最新(2019)首选的spring batch与kafka集成方式是什么？

Question

这是I need current easy to follow instructions for configuring spring integration kafka from XML的后续问题

Spring-integration-kafka在过去的几次迭代中发展了很多，许多旧的例子不再起作用。

特别是，这个从spring-batch世界到spring集成世界的bean将不会实例化，因为KafkaTemplate类没有实现MessagingTemplate。当前推荐的完成此集成的方式是什么？

<bean id="partitionHandler" class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler">
    <property name="stepName" value="fm-step0002.messager"/>
    <property name="gridSize" value="3"/> 
    <property name="messagingOperations" ref="kafkaTemplate"/>
</bean>

下面是我的POM中的一个片段，其中显示了我正在使用的库的版本：

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-core</artifactId>
            <version>5.1.3.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-kafka</artifactId>
            <version>3.1.1.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-file</artifactId>
            <version>5.1.2.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.batch</groupId>
            <artifactId>spring-batch-core</artifactId>
            <version>4.1.1.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.batch</groupId>
            <artifactId>spring-batch-infrastructure</artifactId>
            <version>4.1.1.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.batch</groupId>
            <artifactId>spring-batch-integration</artifactId>
            <version>4.1.1.RELEASE</version>
        </dependency>

Answer 1

MessagingTemplate是Spring Integration的核心组件；它与您正在使用的代理(RabbitMQ、Kafka、JMS等)没有任何关系。

您使用默认通道配置模板(这是kafka出站端点的输入通道)。

请参见batch documentation (单击文档顶部的XML按钮，将示例从java更改为XML)。

这里的示例适用于JMS，但Kafka的配置与此类似。

<bean id="partitionHandler"
   class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler">
  <property name="stepName" value="step1"/>
  <property name="gridSize" value="3"/>
  <property name="replyChannel" ref="outbound-replies"/>
  <property name="messagingOperations">
    <bean class="org.springframework.integration.core.MessagingTemplate">
      <property name="defaultChannel" ref="outbound-requests"/>
      <property name="receiveTimeout" value="100000"/>
    </bean>
  </property>
</bean>

<int:channel id="outbound-requests"/>
<int-jms:outbound-channel-adapter destination="requestsQueue"
    channel="outbound-requests"/>