spring-integration-kafka message-driven-channel-adapter XML 配置

Question

我正在尝试编写一个应用程序来使用来自 Kafka 主题的消息，然后丰富入站消息以添加新的 headers。使用的版本是：

spring-integration-core  -- 5.3.4.RELEASE
spring-integration-kafka -- 3.3.1.RELEASE
spring-boot              -- 2.3.5.RELEASE

在我的集成 XML 中，int-kafka:message-driven-channel-adapter 不允许我提供输出通道或通道属性。我不确定如何将其与流中的下一个通道相关联，该通道可以是 router、transformer、enricher 或任何其他相关内容。

这是一个示例应用程序，它只会从 Kafka 主题接收消息，进行一些转换、丰富等操作，然后将消息的各个部分保存在数据库中。

能否为 multi-channel 集成提供一些 XML 配置模板？

为了完成，这是我正在尝试但无法继续进行的操作：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:int="http://www.springframework.org/schema/integration"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:int-file="http://www.springframework.org/schema/integration/file"
    xmlns:int-ftp="http://www.springframework.org/schema/integration/ftp"
    xmlns:int-xml="http://www.springframework.org/schema/integration/xml"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
    xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
    xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
    xmlns:task="http://www.springframework.org/schema/task"
    xsi:schemaLocation="
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
    http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
    http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd
    http://www.springframework.org/schema/integration/xml http://www.springframework.org/schema/integration/xml/spring-integration-xml.xsd
    http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
    http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
    http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
    http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
    
    <int-kafka:message-driven-channel-adapter
        listener-container="messageListenerContainer"
        send-timeout="5000"
        mode="record"
        retry-template="template"
        recovery-callback="callback"
        error-message-strategy="ems"
        error-channel="errorChannel" />
</beans>

Answer 1

您遇到运行时问题还是 IDE 问题？

有一个测试用例 channel 属性。

https://github.com/spring-projects/spring-integration/blob/15c2d16866a72470b35c65a34e869998f5823f81/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml#L12-L23

我现在在 eclipse 中看到一些虚假错误；调查...

编辑

如果您使用的是 Spring Tool Suite 4，XML 命名空间支持默认处于禁用状态。

启用它解决了我的问题，但我不得不重新启动 IDE。

spring-integration-kafka message-driven-channel-adapter XML 配置

spring-integration-kafka message-driven-channel-adapter XML config

spring-integration

spring-kafka