ToB企服应用市场:ToB评测及商务社交产业平台

标题: kafka怎样包管消息消费的顺序性 [打印本页]

---

作者: 农民    时间: 2025-1-10 06:09
标题: kafka怎样包管消息消费的顺序性

---

一、Kafka 消息顺序性的根本概念

1.1 什么是消息顺序性？

消息顺序性是指在生产者将消息发送到 Kafka 后，消费者按照消息发送的顺序进行消费。

• 全局顺序性：所有分区的消息按顺序消费。
  
• 分区级顺序性：单个分区内的消息按顺序消费。
  

1.2 Kafka 的消息顺序性特点

• 分区级顺序性：Kafka 只能包管同一分区内的消息顺序。
  
• 全局顺序性限制：由于 Kafka 的分区计划，全局顺序消费难以实现。
  

---

二、Kafka 消息顺序性的挑衅

• 分区的并行消费
  每个分区可以被多个消费者并行消费，这大概打乱消息顺序。
  
• 消息重新平衡
  在消费者组发生重新平衡时，大概导致消息的消费顺序庞杂。
  
• 生产端分区策略
  如果消息被分配到不同的分区，无法包管全局顺序。
  
• 消费者失败
  消费者处置惩罚某条消息失败，大概会重新消费消息，打乱顺序。
  

---

三、Kafka 包管消息顺序性的办理方案

3.1 分区级顺序性

Kafka 原生支持分区级顺序性，具体方法：

• 确保同一类消息分配到同一个分区。
  
• 一个分区只允许一个消费者处置惩罚消息。
  

实现方法：

• 使用带有 Key 的消息生产，让 Kafka 按 Key 进行分区。
  
• 设置消费者组，确保每个分区仅由一个消费者处置惩罚。
  

3.2 消费者串行处置惩罚

• 设置消费者的线程数为 1，按顺序处置惩罚消息。
  
• 禁用并行处置惩罚，克制顺序打乱。
  

3.3 分区与业务绑定

将业务 ID 或某些属性作为 Key，确保同一业务的消息落在同一分区，从而包管业务内的顺序性。
3.4 消费失败的处置惩罚策略

• 引入耽误队列或死信队列，克制重新消费打乱顺序。
  
• 使用幂等性处置惩罚，包管每条消息只被处置惩罚一次。
  

---

四、Kafka 消息顺序性实践

以下通过一个示例展示怎样在生产者和消费者中实现分区级的消息顺序消费。
4.1 项目依赖

在 pom.xml 中添加 Kafka 依赖：

1. <dependency>
2.     <groupId>org.apache.kafka</groupId>
3.     <artifactId>kafka-clients</artifactId>
4.     <version>3.5.0</version>
5. </dependency>
6. <dependency>
7.     <groupId>org.springframework.kafka</groupId>
8.     <artifactId>spring-kafka</artifactId>
9.     <version>3.0.0</version>
10. </dependency>

复制代码

---

4.2 设置 Kafka

1. Kafka 设置文件

在 application.yml 中设置 Kafka：

1. spring:
2.   kafka:
3.     bootstrap-servers: localhost:9092
4.     consumer:
5.       group-id: test-group
6.       enable-auto-commit: false
7.       auto-offset-reset: earliest
8.     producer:
9.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
10.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
11.     consumer:
12.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
13.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

复制代码

---

4.3 生产者实现

通过指定消息的 Key，确保同一类型的消息进入同一分区。

1. package com.example.kafka;
3. import org.apache.kafka.clients.producer.KafkaProducer;
4. import org.apache.kafka.clients.producer.ProducerRecord;
5. import org.springframework.stereotype.Service;
7. import java.util.Properties;
9. @Service
10. public class KafkaMessageProducer {
12.     private final KafkaProducer<String, String> producer;
14.     public KafkaMessageProducer() {
15.         Properties props = new Properties();
16.         props.put("bootstrap.servers", "localhost:9092");
17.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
18.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
19.         this.producer = new KafkaProducer<>(props);
20.     }
22.     public void sendMessage(String topic, String key, String value) {
23.         ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
24.         producer.send(record, (metadata, exception) -> {
25.             if (exception != null) {
26.                 System.err.println("Message failed to send: " + exception.getMessage());
27.             } else {
28.                 System.out.println("Message sent to partition " + metadata.partition() + ", offset " + metadata.offset());
29.             }
30.         });
31.     }
32. }

复制代码

---

4.4 消费者实现

通过 @KafkaListener 确保一个分区内只有一个消费者。

1. package com.example.kafka;
3. import org.apache.kafka.clients.consumer.ConsumerRecord;
4. import org.springframework.kafka.annotation.KafkaListener;
5. import org.springframework.stereotype.Service;
7. @Service
8. public class KafkaMessageConsumer {
10.     @KafkaListener(topics = "test-topic", groupId = "test-group", concurrency = "1")
11.     public void consumeMessage(ConsumerRecord<String, String> record) {
12.         System.out.println("Consumed message: Key=" + record.key() + ", Value=" + record.value()
13.                 + ", Partition=" + record.partition() + ", Offset=" + record.offset());
14.         // 模拟顺序处理
15.         processMessage(record.value());
16.     }
18.     private void processMessage(String message) {
19.         System.out.println("Processing message: " + message);
20.     }
21. }

复制代码

---

4.5 测试

1. 发送测试消息

调用生产者发送消息：

1. package com.example.kafka;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.boot.CommandLineRunner;
5. import org.springframework.stereotype.Component;
7. @Component
8. public class KafkaTestRunner implements CommandLineRunner {
10.     @Autowired
11.     private KafkaMessageProducer producer;
13.     @Override
14.     public void run(String... args) throws Exception {
15.         for (int i = 1; i <= 10; i++) {
16.             String key = "user" + (i % 3); // 模拟将相同用户的消息分配到同一分区
17.             String message = "Message " + i;
18.             producer.sendMessage("test-topic", key, message);
19.         }
20.     }
21. }

复制代码

2. 观察消费顺序

运行消费者，输出：

1. Consumed message: Key=user0, Value=Message 3, Partition=2, Offset=0
2. Processing message: Message 3
3. Consumed message: Key=user1, Value=Message 1, Partition=1, Offset=0
4. Processing message: Message 1
5. Consumed message: Key=user2, Value=Message 2, Partition=0, Offset=0
6. Processing message: Message 2
7. ...

复制代码

通过 Key 包管了相同用户的消息进入同一分区，而且按顺序消费。

---

五、Kafka 消息顺序性的留意事项

• 分区数目与消费者数目匹配
  确保每个分区只有一个消费者处置惩罚，克制竞争。
  
• 业务逻辑幂等性
  计划幂等操作，确保即使消息重复消费，也不会影响数据一致性。
  
• 重平衡影响
  只管淘汰消费者组重平衡的频率，克制消费顺序被打乱。
  
• 分区策略优化
  使用自界说分区器，将业务相关的消息路由到相同分区。
  

---

六、总结

Kafka 默认包管分区内的消息顺序，但无法实现全局顺序性。通过合理设置分区策略、确保单分区消费者处置惩罚，以及计划幂等性操作，可以有效实现业务上的消息顺序消费。本文通过 Java 示例展示了怎样利用分区和 Key 来实现分区级的顺序消费。在实际项目中，应结合业务需求权衡性能温顺序性，计划出最优的办理方案。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4