Kafka 对 Produce 与 spring-integration 的确认

Question

使用 Spring Integration Kafka (2.1)，我能够成功地将消息发送到 Kafka 中的主题。

原生 Kafka 客户端 API 提供了成功发送时回调的选项。如何使用 Spring-integration-Kafka 实现相同的目标。下面我的配置和代码供大家参考。

XML 配置

<int:publish-subscribe-channel id="inputToKafka" />

    <int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
                                        auto-startup="true"
                                        channel="inputToKafka"
                                        kafka-template="template"
                                        topic="test"
                                        sync="true">
    </int-kafka:outbound-channel-adapter>

    <bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg>
            <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
                <constructor-arg>
                    <map>
                        <entry key="bootstrap.servers" value="localhost:9092" />
                        <!--<entry key="retries" value="0" />
                        <entry key="batch.size" value="16384" />
                        <entry key="linger.ms" value="0" />
                        <entry key="buffer.memory" value="33554432" /> -->
                        <entry key="key.serializer"
                               value="org.apache.kafka.common.serialization.StringSerializer" />
                        <entry key="value.serializer"
                               value="org.apache.kafka.common.serialization.StringSerializer" />
                    </map>
                </constructor-arg>
            </bean>
        </constructor-arg>
    </bean>

发送消息的Java代码

    @Autowired
    @Qualifier("inputToKafka")
    MessageChannel channel;

channel.send(MessageBuilder.withPayload("Test Message").build());

Answer 1

目前没有“消息”式回调，但您可以使用KafkaTemplate 注册ProducerListener；见KafkaTemplate。