0
0

二、kafka生产与消费全流程

一、使用Java代码生产、消费消息

1、生产者

  1. package com.allwe.client.simple;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.common.serialization.StringSerializer;
  9. import java.util.Properties;
  11. /**
  12.  * kafka生产者配置
  13.  *
  14.  * @Author: AllWe
  15.  * @Date: 2024/09/24/17:57
  16.  */
  17. @Slf4j
  18. public class HelloKafkaProducer {
  19.     public static void main(String[] args) {
  20.         // 设置属性
  21.         Properties properties = new Properties();
  22.         // 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
  23.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  24.         // 设置key和value的序列化器，使java对象转换成二进制数组
  25.         properties.put("key.serializer", StringSerializer.class);
  26.         properties.put("value.serializer", StringSerializer.class);
  28.         // new一个生产者producer
  29.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  30.         try {
  31.             ProducerRecord<String, String> producerRecord;
  32.             try {
  33.                 // 构建消息
  34.                 producerRecord = new ProducerRecord<>("topic_1", "student", "allwe");
  35.                 // 发送消息
  36.                 producer.send(producerRecord);
  37.                 System.out.println("消息发送成功");
  38.             } catch (Exception e) {
  39.                 e.printStackTrace();
  40.             }
  41.         } finally {
  42.             // 释放连接
  43.             producer.close();
  44.         }
  45.     }
  46. }

2、消费者

  1. package com.allwe.client.simple;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.kafka.clients.consumer.ConsumerConfig;
  5. import org.apache.kafka.clients.consumer.ConsumerRecord;
  6. import org.apache.kafka.clients.consumer.ConsumerRecords;
  7. import org.apache.kafka.clients.consumer.KafkaConsumer;
  8. import org.apache.kafka.clients.producer.ProducerConfig;
  9. import org.apache.kafka.common.serialization.StringDeserializer;
  11. import java.time.Duration;
  12. import java.util.Collections;
  13. import java.util.Properties;
  15. /**
  16.  * kafka生产者配置
  17.  *
  18.  * @Author: AllWe
  19.  * @Date: 2024/09/24/17:57
  20.  */
  21. @Slf4j
  22. public class HelloKafkaConsumer {
  23.     public static void main(String[] args) {
  24.         // 设置属性
  25.         Properties properties = new Properties();
  26.         // 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
  27.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  28.         // 设置key和value的序列化器，使java对象转换成二进制数组
  29.         properties.put("key.deserializer", StringDeserializer.class);
  30.         properties.put("value.deserializer", StringDeserializer.class);
  31.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
  33.         // new一个消费者consumer
  34.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
  35.         try {
  36.             // 订阅哪些主题，可以多个，推荐订阅一个主题
  37.             consumer.subscribe(Collections.singleton("topic_1"));
  38.             // 死循环里面实现监听
  39.             while (true) {
  40.                 // 每间隔1s，取一次消息，可能取到多条消息
  41.                 // 设置一秒的超时时间
  42.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  43.                 for (ConsumerRecord<String, String> record : records) {
  44.                     System.out.println("key:" + record.key() + ",value:" + record.value());
  45.                 }
  46.             }
  47.         } finally {
  48.             // 释放连接
  49.             consumer.close();
  50.         }
  51.     }
  52. }

3、踩坑

如果连接的不是本机的kafka，需要在目标机器的kafka配置文件中配置真实的ip地址，如果使用默认的配置或者配置为localhost:9092，kafka.clients会将目标机器的ip解析为127.0.0.1，导致连接不上kafka。

二、生产者

1、序列化器

在上面的demo中，由于消息的key和value都是String类型的，就可以使用kafka.client提供的String序列化器，如果想要发送其他自定义类型的对象，可以手动编写一个序列化器和反序列化器，实现Serializer接口，将对象和byte数组互相转换即可。

需要注意的是，生产者使用的自定义序列化器必须和消费者使用的反序列化器对应，否则无法正确解析消息。

那么什么情况下需要使用自定义序列化器呢？

  1.     -- 需要兼容一些其他协议。

2、分区器

发送的消息被分配到哪个分区中？分区是如何选择的？假设上面的demo中，主题topic_1有4个分区，分别发送4次消息，处理分区的逻辑是怎样的？

这里需要先配置kafka在创建新的主题时，默认的分区数量，我这里配置为了4。

1）指定分区器

可以选择在创建生产者时，给生产者配置相关的分区器，指定具体分区算法。kafka.client提供了一些分区器，或者自己实现一个分区器。

  1. // 设置分区规则
  2. Properties properties = new Properties();
  3. // 1、默认分区器
  4. properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
  5. // 2、统一粘性分区器
  6. properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, UniformStickyPartitioner.class);
  7. // 3、自定义分区器
  8. properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);

自定义分区器：

  1. package com.allwe.client.partitioner;
  3. import org.apache.kafka.clients.producer.Partitioner;
  4. import org.apache.kafka.common.Cluster;
  5. import org.apache.kafka.common.PartitionInfo;
  6. import org.apache.kafka.common.utils.Utils;
  8. import java.util.List;
  9. import java.util.Map;
  11. /**
  12.  * 自定义分区器 - 以value值分区
  13.  */
  14. public class MyPartitioner implements Partitioner {
  15.     @Override
  16.     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
  17.         List<PartitionInfo> partitionInfoList = cluster.partitionsForTopic(topic);
  18.         // 以value值的byte数组处理后再和分区数取模，决定放在哪个分区上
  19.         return Utils.toPositive(Utils.murmur2(valueBytes)) % partitionInfoList.size();
  20.     }
  22.     @Override
  23.     public void close() {
  25.     }
  27.     @Override
  28.     public void configure(Map<String, ?> map) {
  30.     }
  31. }

2）指定分区

也可以选择在构建消息时指定分区，此时的分区优先级最高，不会被其他分区器影响。

  1. # 创建消息时指定分区为 0
  2. ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");

3、生产者发送消息的回调

  1. package com.allwe.client.partitioner;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.clients.producer.RecordMetadata;
  8. import org.apache.kafka.common.serialization.StringSerializer;
  10. import java.util.Properties;
  11. import java.util.concurrent.Future;
  13. /**
  14.  * kafka生产者配置 - 自定义分区器 & 发送消息回调
  15.  *
  16.  * @Author: AllWe
  17.  * @Date: 2024/09/24/17:57
  18.  */
  19. @Slf4j
  20. public class PartitionerProducer {
  21.     public static void main(String[] args) {
  22.         // 设置属性
  23.         Properties properties = new Properties();
  24.         // 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
  25.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  26.         // 设置key和value的序列化器，使java对象转换成二进制数组
  27.         properties.put("key.serializer", StringSerializer.class);
  28.         properties.put("value.serializer", StringSerializer.class);
  29.         // 设置自定义分区器
  30.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
  32.         // new一个生产者producer
  33.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  34.         try {
  35.             ProducerRecord<String, String> producerRecord;
  36.             try {
  37.                 // 构建指定分区的消息，此时指定的分区不会变
  38.                 // producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");
  39.                 for (int i = 0; i < 10; i++) {
  40.                     // 构建消息
  41.                     producerRecord = new ProducerRecord<>("topic_2", "student", "allwe" + i);
  42.                     // 发送消息
  43.                     Future<RecordMetadata> future = producer.send(producerRecord);
  44.                     // 解析回调元数据
  45.                     RecordMetadata recordMetadata = future.get();
  46.                     System.out.println(i + ",offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
  47.                 }
  48.             } catch (Exception e) {
  49.                 e.printStackTrace();
  50.             }
  51.         } finally {
  52.             // 释放连接
  53.             producer.close();
  54.         }
  55.     }
  56. }

打印结果：

4、异步解析生产者发送消息的回调

  1. package com.allwe.client.callBack;
  3. import com.allwe.client.partitioner.MyPartitioner;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.kafka.clients.producer.KafkaProducer;
  6. import org.apache.kafka.clients.producer.ProducerConfig;
  7. import org.apache.kafka.clients.producer.ProducerRecord;
  8. import org.apache.kafka.common.serialization.StringSerializer;
  10. import java.util.Properties;
  12. /**
  13.  * kafka生产者配置 - 异步解析发送消息回调
  14.  *
  15.  * @Author: AllWe
  16.  * @Date: 2024/09/24/17:57
  17.  */
  18. @Slf4j
  19. public class AsynPartitionerProducer {
  20.     public static void main(String[] args) {
  21.         // 设置属性
  22.         Properties properties = new Properties();
  23.         // 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
  24.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  25.         // 设置key和value的序列化器，使java对象转换成二进制数组
  26.         properties.put("key.serializer", StringSerializer.class);
  27.         properties.put("value.serializer", StringSerializer.class);
  28.         // 设置自定义分区器
  29.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
  31.         // new一个生产者producer
  32.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  33.         try {
  34.             ProducerRecord<String, String> producerRecord;
  35.             try {
  36.                 for (int i = 0; i < 10; i++) {
  37.                     // 构建消息
  38.                     producerRecord = new ProducerRecord<>("topic_3", "student", "allwe" + i);
  39.                     // 发送消息, 设置异步回调解析器
  40.                     producer.send(producerRecord, new CallBackImpl());
  41.                 }
  42.                 System.out.println("发送完成,topic_4");
  43.             } catch (Exception e) {
  44.                 e.printStackTrace();
  45.             }
  46.         } finally {
  47.             // 释放连接
  48.             producer.close();
  49.         }
  50.     }
  51. }
  1. package com.allwe.client.callBack;
  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. public class CallBackImpl implements Callback {
  11.     @Override
  12.     public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  13.         if (ObjectUtil.isNull(e)) {
  14.             // 解析回调元数据
  15.             System.out.println("offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
  16.         } else {
  17.             e.printStackTrace();
  18.         }
  19.     }
  20. }

5、生产者缓冲

1）为什么kafka在客户端发送消息的时候需要做一个缓冲？

① 减少IO的开销（单个 -> 批次），需要修改配置文件。

② 减少GC（核心）。

2）如何配置缓冲？

producer.properties配置文件中修改下面两个参数：

消息的大小：batch.size = 默认16384(16K)

暂存的时间：linger.ms = 默认0ms

上面两个条件只要达到一个，就会发送消息，所以在默认配置下，生产一条消息就立即发送。

3）减少GC的原理

producer.properties配置文件的参数：

缓冲池大小：buffer.memory = 默认32M

kafka客户端使用了缓冲池，默认大小32M，当有一条新的消息进入缓冲池，达到了任何一个条件后就发送。发送后不用立即回收内存，而是初始化一下缓冲池即可，减少了GC的次数。

简单说就是利用池化技术减少了对象的创建 -> 减少内存分配次数 -> 减少了垃圾回收次数。

4）使用缓冲池的风险

当缓存的消息超出缓冲池的大小，kafka就会抛出OOM异常。

如果写入消息太快，但是上一次send方法没有执行完，就会导致上一次缓存的消息不能删除，这一次进来的消息又太多，最终写满了缓冲池，触发OOM异常。

解决办法就是适当调整buffer.memory参数和batch.size参数，增加缓冲池大小，缩小每一批次的大小。

三、Kafka Broker

消息从生产者发送出去后，就进入了broker中。在kafka broker中，每一个分区就是一个文件。

四、消费者

1、消费者群组

在消费的过程中，一般情况下使用群组消费，设置group_id_config。

核心：kafka群组消费的负载均衡建立在分区级别。

1）单个群组场景

一个分区只能由一个消费者消费。

在kafka执行过程中，支持动态添加或者减少消费者。

2）多个群组场景

群组之间的消费是互不干扰的，比如群组A的消费者和群组B的消费者可以同时消费同一个分区的消息。

2、Demo记录

写一个生产者，我为了测试顺畅写了一个无限循环的。只启动一次，输入参数即可实现批量发送消息。

  1. package com.allwe.client.singleGroup;
  3. import com.allwe.client.partitioner.MyPartitioner;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.kafka.clients.producer.KafkaProducer;
  6. import org.apache.kafka.clients.producer.ProducerConfig;
  7. import org.apache.kafka.clients.producer.ProducerRecord;
  8. import org.apache.kafka.common.serialization.StringSerializer;
  10. import java.util.Properties;
  11. import java.util.Scanner;
  13. /**
  14.  * kafka生产者配置 - 无限生产消息
  15.  *
  16.  * @Author: AllWe
  17.  * @Date: 2024/09/24/17:57
  18.  */
  19. @Slf4j
  20. public class Producer {
  21.     public static void main(String[] args) {
  22.         // 设置属性
  23.         Properties properties = new Properties();
  24.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  25.         properties.put("key.serializer", StringSerializer.class);
  26.         properties.put("value.serializer", StringSerializer.class);
  27.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
  29.         // new一个生产者producer
  30.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  31.         Scanner scanner = new Scanner(System.in);;
  32.         try {
  33.             int count;
  34.             while (true) {
  35.                 System.out.println("==================输入消息条数===================");
  36.                 String nextLine = scanner.nextLine();
  37.                 if ("exit".equals(nextLine)) {
  38.                     break;
  39.                 }
  40.                 count = Integer.parseInt(nextLine);
  41.                 ProducerRecord<String, String> producerRecord;
  42.                 try {
  43.                     for (int i = 0; i < count; i++) {
  44.                         // 构建消息
  45.                         producerRecord = new ProducerRecord<>("topic_5", "topic_5", "allwe" + i);
  46.                         producer.send(producerRecord);
  47.                     }
  48.                 } catch (Exception e) {
  49.                     e.printStackTrace();
  50.                 }
  51.                 System.out.println("发送完成,topic_5");
  52.             }
  53.         } catch (Exception e) {
  54.             throw new RuntimeException(e);
  55.         } finally {
  56.             // 释放连接
  57.             producer.close();
  58.             scanner.close();
  59.         }
  60.     }
  61. }


生产者控制台展示​​

写一个消费者base类，由于测试消费者需要启动很多类，我这里为了方便写了一个baseConsumer类，调用时new这个类的对象即可调用消费方法。

  1. package com.allwe.client.singleGroup;
  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;
  9. import org.apache.kafka.clients.producer.ProducerConfig;
  10. import org.apache.kafka.common.serialization.StringDeserializer;
  12. import java.time.Duration;
  13. import java.util.Collections;
  14. import java.util.Properties;
  16. /**
  17.  * kafka 消费者配置
  18.  *
  19.  * @Author: AllWe
  20.  * @Date: 2024/09/24/17:57
  21.  */
  22. @Slf4j
  23. @Data
  24. public class SingleGroupBaseConsumer {
  26.     private String groupIdConfig;
  28.     private String topicName;
  30.     private KafkaConsumer<String, String> consumer;
  32.     public SingleGroupBaseConsumer(String groupIdConfig, String topicName) {
  33.         this.groupIdConfig = groupIdConfig;
  34.         this.topicName = topicName;
  35.         createConsumer();
  36.     }
  38.     private void createConsumer() {
  39.         // 设置属性
  40.         Properties properties = new Properties();
  41.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  42.         properties.put("key.deserializer", StringDeserializer.class);
  43.         properties.put("value.deserializer", StringDeserializer.class);
  44.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
  46.         consumer = new KafkaConsumer<>(properties);
  47.     }
  49.     public void poll() {
  50.         try {
  51.             consumer.subscribe(Collections.singleton(topicName));
  52.             while (true) {
  53.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  54.                 int count = 0;
  55.                 for (ConsumerRecord<String, String> record : records) {
  56.                     count = 1;
  57.                     System.out.println("partition:" + record.partition() + ",key:" + record.key() + ",value:" + record.value());
  58.                 }
  59.                 if (count == 1) {
  60.                     // 消费到消息了就打印分隔线
  61.                     System.out.println("===============================");
  62.                 }
  63.             }
  64.         } finally {
  65.             consumer.close();
  66.         }
  67.     }
  68. }

1）单个群组场景

群组id：allwe01

  1. package com.allwe.client.singleGroup;
  3. import lombok.extern.slf4j.Slf4j;
  5. /**
  6.  * kafka消费者启动器
  7.  *
  8.  * @Author: AllWe
  9.  * @Date: 2024/09/24/17:57
  10.  */
  11. @Slf4j
  12. public class SingleGroupConsumer_1 {
  13.     public static void main(String[] args) {
  14.         SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe01", "topic_5");
  15.         singleGroupBaseConsumer.poll();
  16.     }
  17. }


消费者控制台展示

我这里只放了一个消费者的消费记录，根据消费者控制台打印的数据，可以看到两条信息：

① 该消费者只能消费分区=1的消息。

② 消费者消费消息时，每次拿到的消息数量不确定。

2）多个群组场景

群组id：allwe02

  1. package com.allwe.client.group;
  3. import com.allwe.client.singleGroup.SingleGroupBaseConsumer;
  4. import lombok.extern.slf4j.Slf4j;
  6. /**
  7.  * kafka消费者启动器
  8.  *
  9.  * @Author: AllWe
  10.  * @Date: 2024/09/24/17:57
  11.  */
  12. @Slf4j
  13. public class GroupConsumer_1 {
  14.     public static void main(String[] args) {
  15.         SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe02", "topic_5");
  16.         singleGroupBaseConsumer.poll();
  17.     }
  18. }


消费者控制台展示

可以看到，这里新加入了一个消费者群组，只有一个消费者，它就消费到了全部分区的消息。

3、ACK确认

消费者在成功消费消息后，会进行ACK确认。提交最后一次消费消息的偏移量，下一次消费就从上次提交的偏移量开始，如果一个新的消费者群组消费一个主题的消息，可以根据不同的配置来指定起始的偏移量。

  1. // 从最早的消息开始消费
  2. properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  4. // 从已提交的偏移量开始消费 - 默认配置
  5. properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

在kafka内部，有一个名字叫【__consumer_offsets】的主题，保存了消费者对各个主题的消费偏移量。消费者每一次发送的ACK确认，都会更新这个主题中的偏移量数据。

1）自动提交ACK的消费模式

默认的消费模式。

只要拿到了消息，就自动提交ACK确认。

但是有一个风险，就是虽然消费者成功取到了消息，但是在程序处理过程中出现了异常，同时提交了ACK确认，那么这条消息就永远不会被正确地处理。

所以有时候我们需要避免自动提交ACK确认，改成手动提交ACK确认。

2）手动提交ACK确认

取消自动提交

  1. // 取消自动提交
  2. properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
① 同步提交
  1. // 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行
  2. consumer.commitSync();

立刻进行ACK确认。但是容易造成阻塞，只有等待ACK确认成功后，才会继续执行程序。如果ACK确认不成功，就会一直重试。

② 异步提交
  1. // 异步提交ACK确认
  2. consumer.commitAsync();

异步提交不会阻塞应用程序，提交失败不会重试提交。

③ 组合使用demo
  1.     public void poll() {
  2.         try {
  3.             consumer.subscribe(Collections.singleton(topicName));
  4.             while (true) {
  5.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  6.                 int count = 0;
  7.                 for (ConsumerRecord<String, String> record : records) {
  8.                     count = 1;
  9.                     System.out.println("partition:" + record.partition() + ",offset:" + record.offset() +",key:" + record.key() + ",value:" + record.value());
  10.                 }
  11.                 if (count == 1) {
  12.                     // 消费到消息了就打印分隔线
  13.                     System.out.println("===============================");
  14.                 }
  15.                 // 异步提交ACK确认
  16.                 consumer.commitAsync();
  17.             }
  18.         } finally {
  19.             try {
  20.                 // 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行
  21.                 consumer.commitSync();
  22.             } finally {
  23.                 consumer.close();
  24.             }
  25.         }
  26.     }

3）手动批量提交ACK确认

如果消费者在某一时刻取到的消息数量太多，那么给每一条消息单独提交ACK确认太浪费资源，可以选择批量提交ACK确认。核心思想就是在程序中暂存偏移量，达到设定的阈值后就触发批量提交。

kafka.Consumer提供的异步提交ACK方法支持批量提交。

五、多线程安全问题

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);
  31.     public static void main(String[] args) {
  32.         // 消费者配置
  33.         Properties properties = new Properties();
  34.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
  35.         properties.put("key.deserializer", StringDeserializer.class);
  36.         properties.put("value.deserializer", StringDeserializer.class);
  37.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从头开始消费
  39.         for (Integer i = 0; i < THREAD_COUNT; i++) {
  40.             executorService.submit(new ConcurrentConsumer(properties, "allwe01", "topic_6"));
  41.         }
  42.     }
  43. }

六、群组协调

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));
标签： kafka
本文转载自: https://blog.csdn.net/weixin_47201257/article/details/142527806
版权归原作者 AllWe$ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“二、kafka生产与消费全流程”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？