Spring Integration for Kafka多主题配置：动态发送消息至不同主题

刚好我之前也处理过类似的需求，用Spring Integration Kafka实现根据场景发送到不同主题，结合你目前用XML配置int-kafka:producer-configuration的情况，给你两种实用的实现方案：

方案一：通过消息头动态指定主题 #

这种方式最灵活，不需要额外配置多个Producer，只需要在发送消息时通过消息头指定目标主题，Producer会自动读取这个头的值作为发送的主题。

XML配置修改 #

把原来固定的topic属性改成SpEL表达式，读取消息头里的主题值：

<int-kafka:producer-configuration id="kafkaProducerConfig"
                                  bootstrap-servers="localhost:9092"
                                  key-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  value-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  topic="#{headers['target_topic']}"/> <!-- 用SpEL读取消息头中的主题 -->

<!-- 定义输出通道，绑定到Producer适配器 -->
<int:channel id="kafkaOutputChannel"/>
<int-kafka:outbound-channel-adapter id="kafkaOutboundAdapter"
                                    channel="kafkaOutputChannel"
                                    producer-configuration="kafkaProducerConfig"/>

Java代码发送示例 #

发送消息时，在Message中添加target_topic消息头，指定要发送的主题：

@Autowired
private MessageChannel kafkaOutputChannel;

public void sendMessageToDynamicTopic(String content, String targetTopic) {
    // 构建带主题头的消息
    Message<String> message = MessageBuilder.withPayload(content)
            .setHeader("target_topic", targetTopic)
            .build();
    kafkaOutputChannel.send(message);
}

// 调用示例：分别发送到topic1和topic2
sendMessageToDynamicTopic("消息内容1", "topic1");
sendMessageToDynamicTopic("消息内容2", "topic2");

如果想用Kafka官方规范的消息头常量，可以改用org.springframework.kafka.support.KafkaHeaders.TOPIC：
代码里改成setHeader(KafkaHeaders.TOPIC, targetTopic)，XML里的表达式改成#{headers[T(org.springframework.kafka.support.KafkaHeaders).TOPIC]}，这样更符合框架规范。

方案二：基于路由配置多Producer实例 #

如果你的发送场景是固定的（比如只有两种主题需要区分），可以配置多个Producer分别对应不同主题，然后通过路由器（Router）根据业务条件选择发送到对应的Producer通道。

XML配置 #

<!-- 第一个Producer配置，对应topic1 -->
<int-kafka:producer-configuration id="kafkaProducerConfig1"
                                  bootstrap-servers="localhost:9092"
                                  key-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  value-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  topic="topic1"/>

<!-- 第二个Producer配置，对应topic2 -->
<int-kafka:producer-configuration id="kafkaProducerConfig2"
                                  bootstrap-servers="localhost:9092"
                                  key-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  value-serializer="org.apache.kafka.common.serialization.StringSerializer"
                                  topic="topic2"/>

<!-- 定义两个输出通道，分别绑定到不同Producer -->
<int:channel id="kafkaOutputChannel1"/>
<int-kafka:outbound-channel-adapter id="kafkaOutboundAdapter1"
                                    channel="kafkaOutputChannel1"
                                    producer-configuration="kafkaProducerConfig1"/>

<int:channel id="kafkaOutputChannel2"/>
<int-kafka:outbound-channel-adapter id="kafkaOutboundAdapter2"
                                    channel="kafkaOutputChannel2"
                                    producer-configuration="kafkaProducerConfig2"/>

<!-- 配置路由器，根据消息头选择目标通道 -->
<int:router input-channel="kafkaRouterChannel" expression="#{headers['scene_type']}">
    <int:mapping value="SCENE_A" channel="kafkaOutputChannel1"/>
    <int:mapping value="SCENE_B" channel="kafkaOutputChannel2"/>
</int:router>

<!-- 路由器的输入通道，业务代码发送到这个通道即可 -->
<int:channel id="kafkaRouterChannel"/>

Java代码发送示例 #

发送时通过scene_type消息头指定场景，路由器会自动路由到对应主题的Producer：

@Autowired
private MessageChannel kafkaRouterChannel;

public void sendMessageByScene(String content, String sceneType) {
    Message<String> message = MessageBuilder.withPayload(content)
            .setHeader("scene_type", sceneType)
            .build();
    kafkaRouterChannel.send(message);
}

// 调用示例：场景A发送到topic1，场景B发送到topic2
sendMessageByScene("场景A的消息", "SCENE_A");
sendMessageByScene("场景B的消息", "SCENE_B");

补充说明 #

• 如果路由条件是基于消息内容（比如payload对象的某个字段），可以把路由器的expression改成payload.sceneType（假设payload是包含sceneType属性的自定义对象）。
• 方案一更适合主题不固定的动态场景，方案二更适合固定场景的区分，维护起来更清晰。

内容的提问来源于stack exchange，提问作者KVNS_JavaLearner