三、kafka消费的全流程

五、多线程安全问题

1、多线程安全的定义

使用多线程访问一个资源，这个资源始终都能体现出正确的行为。
不被运行的环境影响、多线程可以交替访问、不需要任何额外的同步和协同。
2、Java实现多线程安全生产者

   这里只是模仿多线程环境下使用生产者发送消息，其实没有做额外的线程安全操纵，就是把生产者当成了一个公共资源，全部线程都可以访问这个生产者。
  kafka默认客户端提供的生产者本身就是线程安全的，因为生产者发送消息只有一步操纵，就是发送消息。只要消息进入消息缓冲区就可以发送给broker，不会出现消息重复发送。

1. package com.allwe.client.concurrent;
3. import com.allwe.client.partitioner.MyPartitioner;
4. import lombok.Data;
5. import lombok.extern.slf4j.Slf4j;
6. import org.apache.kafka.clients.producer.KafkaProducer;
7. import org.apache.kafka.clients.producer.ProducerConfig;
8. import org.apache.kafka.clients.producer.ProducerRecord;
9. import org.apache.kafka.common.serialization.StringSerializer;
11. import java.util.Properties;
12. import java.util.concurrent.CountDownLatch;
13. import java.util.concurrent.ExecutorService;
14. import java.util.concurrent.Executors;
16. /**
17. * 生产者多线程安全 - 测试demo
18. *
19. * @Author: AllWe
20. * @Date: 2024/09/27/9:30
21. */
22. @Data
23. @Slf4j
24. public class ConcurrentProducerWorker {
26.     /**
27.      * 消息数量
28.      */
29.     private static final int RECORD_COUNT = 1000;
31.     /**
32.      * 固定线程池 - 线程数等于CPU核数
33.      */
34.     private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
36.     /**
37.      * 发令枪
38.      */
39.     private static final CountDownLatch countDownLatch = new CountDownLatch(RECORD_COUNT);
41.     /**
42.      * 生产者 - 这里让所有的线程都共享同一个生产者
43.      */
44.     private static KafkaProducer<String, String> kafkaProducer;
46.     /**
47.      * 类初始化的时候 - 创建生产者实例
48.      */
49.     static {
50.         // 设置属性
51.         Properties properties = new Properties();
52.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
53.         properties.put("key.serializer", StringSerializer.class);
54.         properties.put("value.serializer", StringSerializer.class);
55.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
57.         kafkaProducer = new KafkaProducer<>(properties);
58.     }
60.     /**
61.      * 启动器
62.      */
63.     public static void main(String[] args) {
64.         try {
65.             // 循环创建消息
66.             for (int count = 0; count < RECORD_COUNT; count++) {
67.                 ProducerRecord<String, String> record = new ProducerRecord<>("topic_6", "allwe", "allwe_" + count);
68.                 executorService.submit(new ConcurrentProducer(record, kafkaProducer, countDownLatch));
69.             }
70.             countDownLatch.await();
71.         } catch (Exception e) {
72.             e.printStackTrace();
73.         } finally {
74.             // 关闭生产者连接
75.             kafkaProducer.close();
76.             // 释放线程池资源
77.             executorService.shutdown();
78.         }
79.     }
80. }

复制代码

1. package com.allwe.client.concurrent;
3. import lombok.Data;
4. import lombok.extern.slf4j.Slf4j;
5. import org.apache.kafka.clients.producer.KafkaProducer;
6. import org.apache.kafka.clients.producer.ProducerRecord;
8. import java.util.concurrent.CountDownLatch;
10. /**
11. * 生产者多线程安全 - 测试demo
12. *
13. * @Author: AllWe
14. * @Date: 2024/09/27/9:30
15. */
16. @Data
17. @Slf4j
18. public class ConcurrentProducer implements Runnable {
20.     /**
21.      * 消息体
22.      */
23.     private ProducerRecord<String, String> record;
25.     /**
26.      * 生产者
27.      */
28.     private KafkaProducer<String, String> producer;
30.     /**
31.      * 发令枪
32.      */
33.     private CountDownLatch countDownLatch;
35.     public ConcurrentProducer(ProducerRecord<String, String> record, KafkaProducer<String, String> producer, CountDownLatch countDownLatch) {
36.         this.record = record;
37.         this.producer = producer;
38.         this.countDownLatch = countDownLatch;
39.     }
41.     @Override
42.     public void run() {
43.         try {
44.             String name = Thread.currentThread().getName();
45.             producer.send(record, new ConcurrentCallBackImpl(name));
46.             countDownLatch.countDown();
47.         } catch (Exception e) {
48.             e.printStackTrace();
49.         }
50.     }
51. }

复制代码

1. package com.allwe.client.concurrent;
3. import cn.hutool.core.util.ObjectUtil;
4. import org.apache.kafka.clients.producer.Callback;
5. import org.apache.kafka.clients.producer.RecordMetadata;
7. /**
8. * 异步发送消息回调解析器
9. *
10. * @Author: AllWe
11. * @Date: 2024/09/27/9:30
12. */
13. public class ConcurrentCallBackImpl implements Callback {
15.     private String threadName;
17.     public ConcurrentCallBackImpl(String threadName) {
18.         this.threadName = threadName;
19.     }
21.     @Override
22.     public void onCompletion(RecordMetadata recordMetadata, Exception e) {
23.         if (ObjectUtil.isNull(e)) {
24.             // 解析回调元数据
25.             System.out.println(threadName + "|-offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
26.         } else {
27.             e.printStackTrace();
28.         }
29.     }
30. }

复制代码

3、Java实现多线程安全消费者

   kafka客户端提供的消费者不是多线程安全的，是因为消费者在消费消息的时候，需要有2步操纵：取消息和ACK确认，在多线程场景下可能会出现：
  ① 线程1取到了消息，但是没来得及进行ACK确认。
  ② 线程2进来了，又消费了一次相同的消息。
  ③ 线程2提交ACK确认。
  ④ 线程1提交ACK确认。
  如许就会产生重复消费，这个时候就需要对消费者进行额外处理。
  有两个处理方案：
  ① 给消费过程加锁，但是会降低程序执行效率。
  ② 每一个线程都创建本身的消费者，只消费本身分区内的数据。
  我写的demo是使用第二种办法。

1. package com.allwe.client.concurrent;
3. import lombok.Data;
4. import lombok.extern.slf4j.Slf4j;
5. import org.apache.kafka.clients.consumer.ConsumerConfig;
6. import org.apache.kafka.clients.consumer.ConsumerRecord;
7. import org.apache.kafka.clients.consumer.ConsumerRecords;
8. import org.apache.kafka.clients.consumer.KafkaConsumer;
10. import java.time.Duration;
11. import java.util.Collections;
12. import java.util.Properties;
14. /**
15. * 线程安全消费者 - 测试demo
16. *
17. * @Author: AllWe
18. * @Date: 2024/09/27/12:19
19. */
20. @Data
21. @Slf4j
22. public class ConcurrentConsumer implements Runnable {
23.     /**
24.      * 消费者配置参数
25.      */
26.     private Properties properties;
28.     /**
29.      * 群组id
30.      */
31.     private String groupId;
33.     /**
34.      * 消费主题
35.      */
36.     private String topicName;
38.     /**
39.      * 消费者实例
40.      */
41.     private KafkaConsumer<String, String> consumer;
43.     public ConcurrentConsumer(Properties properties, String groupId, String topicName) {
44.         this.properties = properties;
45.         this.groupId = groupId;
46.         this.topicName = topicName;
48.         // 补充配置参数
49.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
50.         // 创建消费者实例 - 每一个线程都创建自己的消费者,避免共享相同的消费者实例
51.         consumer = new KafkaConsumer<>(properties);
52.         // 配置消费主题
53.         consumer.subscribe(Collections.singleton(topicName));
54.     }
56.     @Override
57.     public void run() {
58.         try {
59.             String threadName = Thread.currentThread().getName();
60.             while (true) {
61.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
62.                 for (ConsumerRecord<String, String> record : records) {
63.                     StringBuilder stringBuilder = new StringBuilder(threadName).append("|-");
64.                     stringBuilder.append("partition:").append(record.partition());
65.                     stringBuilder.append("offset:").append(record.offset());
66.                     stringBuilder.append("key:").append(record.key());
67.                     stringBuilder.append("value:").append(record.value());
69.                     System.out.println(stringBuilder);
70.                 }
71.             }
72.         } finally {
73.             consumer.close();
74.         }
75.     }
76. }

复制代码

1. package com.allwe.client.concurrent;
3. import lombok.Data;
4. import lombok.extern.slf4j.Slf4j;
5. import org.apache.kafka.clients.consumer.ConsumerConfig;
6. import org.apache.kafka.common.serialization.StringDeserializer;
8. import java.util.Properties;
9. import java.util.concurrent.ExecutorService;
10. import java.util.concurrent.Executors;
12. /**
13. * 多线程安全消费者 - 测试demo
14. *
15. * @Author: AllWe
16. * @Date: 2024/09/27/12:34
17. */
18. @Data
19. @Slf4j
20. public class ConcurrentConsumerWorker {
21.     /**
22.      * 消费线程数
23.      */
24.     private static final Integer THREAD_COUNT = 2;
26.     /**
27.      * 线程池 - 2个线程,别超过目标主题的分区数
28.      */
29.     private static ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
32.     public static void main(String[] args) {
33.         // 消费者配置
34.         Properties properties = new Properties();
35.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
36.         properties.put("key.deserializer", StringDeserializer.class);
37.         properties.put("value.deserializer", StringDeserializer.class);
38.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从头开始消费
40.         for (Integer i = 0; i < THREAD_COUNT; i++) {
41.             executorService.submit(new ConcurrentConsumer(properties, "allwe01", "topic_6"));
42.         }
43.     }
44. }

复制代码

六、群组协调

1、群主

在每一个群组内部，都有一个【群主】。往往是第一个注册进入群组的消费者承担，它的职责是读取当前群组消费的主题，以及目标主题的分区信息。
群主节点的数据权限高于平凡消费者，它可以获取全部消费者节点对应的分区信息。但是平凡消费者节点只能看见本节点的分区信息。
2、消费者协调器

属于客户端，每个消费者群组内部都有一个消费者协调器，用于获取群主节点保存的分区信息，再协调群组内的其他消费者处理哪些主题和分区。
分配好使命后将配置信息推送给【组协调器】，组协调器再将消息发送给不同的消费者。
当群组内出现某个节点掉线、上线时，消费者协调器也会参与协调。
   1、向【组协调器】发送入组请求。
  2、发起同步组的请求 -- 由群组计算分配策略，确定消费者的分区分别，发送给组协调器。
  3、心跳机制（与组协调器维持）。
  4、提交ACK确认（发起已经提交的消费偏移量的请求）。
  5、主动发起离组请求。
  3、组协调器

属于kafka broker，重要负责以下功能：
   1、处理申请加入群组的消费者，而且选举群主。
  2、收到同步组的请求后，触发分区再平衡，同步新的分配方案。
  3、心跳机制（与客户端维持），如果得知哪些客户端掉线了，触发分区再平衡机制。
  4、管理消费者已经消费的偏移量，保存在主题【__consumer_offsets】，默认有50个分区。
  4、新的消费者加入群组的处理流程

   1、消费者客户端启动、重连，都会给组协调器发送一个入组请求（joinGroup请求）。
  2、消费者客户端完成joinGroup后，消费者协调器向组协调器发起同步组请求（SyncGroup请求），获取新的分配方案。
  3、入组后保持心跳（客户端控制参数：max.poll.interval.ms）。
  4、消费者客户端掉线，触发离组处理。
  5、消费者群组的信息存储在哪里

存储在__consumer_offsets文件中，groupName.hashCode() % 50，获取配置文件的编号。
七、分区再平衡

1、功能

   针对单个消费者群组，对群组内的消费者负责的分区进行重新分配。
  1、假设【主题α】有三个分区，分别是①、②、③。
  2、进来两个消费者A、B。A负责分区①，B负责分区②③。
  3、又进来一个消费者C，再平衡监听器就把分区③分配给C。
  4、消费者C掉线，再平衡监听器把分区③分配给A或者B。
  2、Java代码验证分区再平衡

1. package com.allwe.client.reBalance;
3. import lombok.Data;
4. import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
5. import org.apache.kafka.clients.consumer.KafkaConsumer;
6. import org.apache.kafka.clients.consumer.OffsetAndMetadata;
7. import org.apache.kafka.common.TopicPartition;
9. import java.util.Collection;
10. import java.util.Map;
11. import java.util.concurrent.ConcurrentHashMap;
13. /**
14. * 分区再均衡处理器
15. *
16. * @Author: AllWe
17. * @Date: 2024/10/17/8:05
18. */
19. @Data
20. public class ReBalanceHandler implements ConsumerRebalanceListener {
22.     // 记录每个分区的消费偏移量
23.     public final static ConcurrentHashMap<TopicPartition, Long> partitionOffsetMap = new ConcurrentHashMap<TopicPartition, Long>();
25.     private final Map<TopicPartition, OffsetAndMetadata> currOffsets;
27.     private final KafkaConsumer<String, String> consumer;
29.     public ReBalanceHandler(Map<TopicPartition, OffsetAndMetadata> currOffsets, KafkaConsumer<String, String> consumer) {
30.         this.currOffsets = currOffsets;
31.         this.consumer = consumer;
32.     }
34.     // 分区再均衡之前
35.     // 某一个消费者在让出分区之前，需要先将已消费的偏移量提交
36.     @Override
37.     public void onPartitionsRevoked(Collection<TopicPartition> collection) {
38.         // 线程id
39.         final String id = Thread.currentThread().getId() + "";
40.         System.out.println(id + "-onPartitionsRevoked参数值为：" + collection);
41.         System.out.println(id + "-服务器准备分区再均衡，提交偏移量。当前偏移量为:" + currOffsets);
42.         //我们可以不使用consumer.commitSync(currOffsets);
43.         //提交偏移量到kafka,由我们自己维护*/
44.         //开始事务
45.         //偏移量写入数据库
46.         System.out.println("分区偏移量表中:" + partitionOffsetMap);
47.         for (TopicPartition topicPartition : collection) {
48.             partitionOffsetMap.put(topicPartition, currOffsets.get(topicPartition).offset());
49.         }
50.         // 同步提交偏移量，等到成功后再往后执行
51.         consumer.commitSync(currOffsets);
52.     }
54.     // 分区再均衡之后
55.     // 新的消费者接管分区后，从上一次的偏移量开始消费
56.     @Override
57.     public void onPartitionsAssigned(Collection<TopicPartition> collection) {
58.         // 线程id
59.         final String threadId = Thread.currentThread().getId() + "";
60.         System.out.println(threadId + "|-再均衡完成，onPartitionsAssigned参数值为：" + collection);
61.         System.out.println("分区偏移量表中：" + partitionOffsetMap);
63.         for (TopicPartition topicPartition : collection) {
64.             System.out.println(threadId + "-topicPartition" + topicPartition);
65.             // 取得接管分区之前的偏移量
66.             Long offset = partitionOffsetMap.get(topicPartition);
67.             if (offset == null) continue;
68.             consumer.seek(topicPartition, partitionOffsetMap.get(topicPartition));
69.         }
70.     }
72.     @Override
73.     public void onPartitionsLost(Collection<TopicPartition> partitions) {
74.         ConsumerRebalanceListener.super.onPartitionsLost(partitions);
75.     }
76. }

复制代码

1. package com.allwe.client.reBalance;
3. import lombok.Data;
4. import lombok.extern.slf4j.Slf4j;
5. import org.apache.kafka.clients.consumer.*;
6. import org.apache.kafka.common.TopicPartition;
8. import java.time.Duration;
9. import java.util.Collections;
10. import java.util.HashMap;
11. import java.util.Map;
12. import java.util.Properties;
14. /**
15. * 线程安全消费者 - 测试demo
16. *
17. * @Author: AllWe
18. * @Date: 2024/09/27/12:19
19. */
20. @Data
21. @Slf4j
22. public class ConcurrentConsumer implements Runnable {
23.     /**
24.      * 消费者配置参数
25.      */
26.     private Properties properties;
28.     /**
29.      * 群组id
30.      */
31.     private String groupId;
33.     /**
34.      * 消费主题
35.      */
36.     private String topicName;
38.     /**
39.      * 消费者实例
40.      */
41.     private KafkaConsumer<String, String> consumer;
43.     /**
44.      * 记录分区消费者偏移量
45.      */
46.     private final Map<TopicPartition, OffsetAndMetadata> currOffsets = new HashMap<>();
48.     public ConcurrentConsumer(Properties properties, String groupId, String topicName) {
49.         this.properties = properties;
50.         this.groupId = groupId;
51.         this.topicName = topicName;
53.         // 补充配置参数
54.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
55.         // 创建消费者实例 - 每一个线程都创建自己的消费者,避免共享相同的消费者实例
56.         consumer = new KafkaConsumer<>(properties);
57.         // 配置消费主题 - 配置再均衡监听器
58.         consumer.subscribe(Collections.singleton(topicName), new ReBalanceHandler(currOffsets,consumer));
59.     }
61.     @Override
62.     public void run() {
63.         try {
64.             String threadName = Thread.currentThread().getName();
65.             Integer offset = 0;
66.             while (true) {
67.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
68.                 for (ConsumerRecord<String, String> record : records) {
69.                     StringBuilder stringBuilder = new StringBuilder(threadName).append("|-");
70.                     stringBuilder.append("partition:").append(record.partition());
71.                     stringBuilder.append(",offset:").append(record.offset());
72.                     stringBuilder.append(",key:").append(record.key());
73.                     stringBuilder.append(",value:").append(record.value());
75.                     System.out.println(stringBuilder);
77.                     offset++;
78.                     currOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(offset, "no"));
79.                 }
80.             }
81.         } finally {
82.             consumer.close();
83.         }
84.     }
85. }

复制代码

自定义一个再平衡监听器，消费者在订阅接口中指定这个监听器，即可自动执行监听器的使命。

1. // 配置消费主题 - 配置再均衡监听器
2. consumer.subscribe(Collections.singleton(topicName), new ReBalanceHandler(currOffsets,consumer));

复制代码

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

继续阅读请点击广告