kafka消息监听

1，
spring配置kafka网址

2，listener

1.  
   @Component
2.  
   public class OrderMsgListener {
3.  
    
4.  
   @KafkaListener(topics = "order",groupId = "order-service")
5.  
   public void listen(ConsumerRecord record){
6.  
   System.out.println("收到消息：" record); //可以监听到发给kafka的新消息，以前的拿不到
7.  
   }
8.  
    
9.  
   @KafkaListener(groupId = "order-service-2",topicPartitions = {
10.  
    @TopicPartition(topic = "order",partitionOffsets = {
11.  
    @PartitionOffset(partition = "0",initialOffset = "0")
12.  
    })
13.  
    })//这一块注解表示可以监听kafka之前的消息
14.  
    public void listenAll(ConsumerRecord record){
15.  
    System.out.println("收到partion-0消息：" record);
16.  
    }
17.  
    }

groupId表示分组，不同组的消费者不是竞争关系

3，

1.  
   @KafkaListener(groupId = "order-service-2",topicPartitions = {
2.  
   @TopicPartition(topic = "order",partitionOffsets = {
3.  
   @PartitionOffset(partition = "0",initialOffset = "0")
4.  
   })
5.  
   })

这段代码使用了Spring Kafka提供的注解@KafkaListener来定义一个Kafka消费者。具体的配置如下：

1. groupId = "order-service-2"：指定该消费者所属的消费者组ID，即"order-service-2"。
2. topicPartitions：表示要订阅的主题和分区信息，是一个数组。
3. @TopicPartition(topic = "order", partitionOffsets = { ... })：表示订阅名为"order"主题的特定分区。
4. @PartitionOffset(partition = "0", initialOffset = "0")：表示订阅的分区为0，并设置初始偏移量为0。

这段代码将创建一个Kafka消费者，用于订阅名为"order"的Kafka主题下的0号分区，并从初始偏移量0开始消费消息。

在实际应用中，您可能会根据需要添加其他的@TopicPartition和@PartitionOffset注解可以订阅多个主题和分区，并指定每个分区的初始偏移量。

值得注意的是，上述代码是使用Spring Kafka提供的注解方式来创建Kafka消费者。通过该注解，您可以方便地定义多个消费者，并且框架会自动处理与Kafka的连接、消费消息等底层细节。

4，

 主题：topics；消息是发送给某个主题

这篇好文章是转载于：学新通技术网