基于Spring Integration的Apache Kafka JMS消费者

• Kafka没有跟踪哪些消息被消费而言，Apache Kafka是无状态的。这是JMS消费者的责任。这种简单性使Kafka非常快。
• Apache Kafka消息的JMS消费者者需要保持消息偏移以便能够跟踪他消费使用的消息。偏移也使他有机会回到原来特定位置并再次重新阅读消息。
• 在将消息写入Kafka时，消息将写入分区的末尾。这样可以节省磁盘，这也是Apache Kafka的另一个性能提升。

使用Spring Integration编写Apache Kafka的JMS消费者，有两种选择：

• 使用基于高级 Apache Kafka API的入站通道Inbound Channel 适配器。
• 使用基于Apache Kafka的SimpleConsumer的消息驱动适配器。

另一方面，使用Kafka的SimpleConsumer的消息驱动适配器为我们提供了org.springframework.integration.kafka.listener.OffsetManager来跟踪偏移量。

在本演示中，我们将测试通过InboundChannel Adapter获取消息。因为大多数时候这个选项就足够了。

我们将构建简单的基于Spring-Boot的应用程序，并集成了Spring Integration依赖项。主类看起来像这样：

@SpringBootApplication
@EnableIntegration
@ImportResource("applicationContext.xml")
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

Spring集成部分：

<int:channel id="inputFromKafka">
        <int:queue />
    </int:channel>

    <bean id="messageProcessor" class="com.example.processor.MessageProcessor" />

    <int:service-activator input-channel="inputFromKafka"
        ref="messageProcessor">
        <int:poller fixed-delay="10" />
    </int:service-activator>

我们在这里简单地定义了queue-pollable channel inputFromKafka，然后将该通道中的消息移交给com.example.processor.MessageProcessor。轮询间隔为10毫秒。在这个处理器中，让我们打印出消息。

public class MessageProcessor {
   public void handleMessage(Message<?> message) {
         System.out.println("<strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong>");
System.out.println("Spring integration received:"+message.getPayload());
System.out.println("<strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong><strong>*</strong>");
    }
}

非常重要的是要记住，来自Apache Kafka JMS消息转换的Spring Integration消息的有效负载结构如下：

Map>>

第一个映射的键是主题的ID，第二个映射的键是主题分区的ID，List