二、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数组相互转换即可。
需要注意的是，生产者使用的自界说序列化器必须和消费者使用的反序列化器对应，否则无法精确解析消息。
那么什么情况下需要使用自界说序列化器呢？
        -- 需要兼容一些其他协议。
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方法支持批量提交。

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