0
0

详细讲解如何使用Java连接Kafka构建生产者和消费者（带测试样例）

1 缘起

学习消息队列的过程中，先补习了RabbitMQ相关知识，
接着又重温了Kafka相关的知识，
发现，我并没有积累Java原生操作Kafka的文章，
只使用SpringBoot集成过Kafka，
所以，本次是纯Java的方式操作Kafka，
构建生产者和消费者，本地部署Kafka环境，给出测试样例的测试结果，
同时，讲解部分通用的参数，
及给出通过命令行启动生产者和消费者的测试样例，
分享如下，帮助读者学习Kafka基础操作。

2 环境准备

下载kafka：https://download.csdn.net/download/Xin_101/19787459

2.1 启动zookeeper

  1. bin/zookeeper-server-start.sh config/zookeeper.properties

2.2 启动kafka

  1. bin/kafka-server-start.sh config/server.properties

2.3 新建topic

  1. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic my-topic

在这里插入图片描述

2.4 依赖

  1. <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.2.0</version></dependency>

3 Kafka通用配置

Kafka的生产者和消费者参数比较多，这里仅列出一些测试用的参数，
区分生产者和消费者。

3.1 生产者配置参数

序号参数描述1bootstrap.serversKafka主机2acks生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化
消息确认：
0：生产者不等待服务器确认，此时retry参数不生效
1：leader写入记录到log，不会等待follower的确认即向生产者发送通知
all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为-13retries生产者重试次数4batch.size生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes=16K5linger.ms生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒6buffer.memory生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes=32M7key.serializer生产者：键编码器8value.serializer生产者：值编码器

3.2 消费者配置参数

序号参数描述1bootstrap.serversKafka主机2group.id消费者：消费topic的组ID3enable.auto.commit消费者：后台定期提交offset4auto.commit.interval.ms消费者提交offset的时间间隔：单位：毫秒，当enable.auto.commit为true时生效5auto.offset.reset消费者：重启后配置offset
earliest：消费者恢复到当前topic最早的offset
latest：消费者从最新的offset开始消费
none：如果消费者组没找到之前的offset抛出异常
其他任何值都会抛出异常6key.deserializer消费者：键解码器7value.deserializer消费者：值解码器

3.3 Kafka通用参数封装

由于参数众多，这里封装了一个Kafka通用参数类，给了默认值，
本地测试，直接使用默认参数，
同时给出了有参构造器，自定义参数，
代码样例如下。

  1. packagecom.monkey.java_study.mq.kafka;importjava.util.Collection;importjava.util.Collections;/**
  2.  * Kafka通用配置.
  3.  *
  4.  * @author xindaqi
  5.  * @since 2022-08-03 9:49
  6.  */publicclassKafkaCommonProperties{/**
  7.      * Kafka主机
  8.      */privateString kafkaHost ="192.168.211.129:9092";/**
  9.      * 生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化
  10.      * 消息确认：
  11.      * 0：生产者不等待服务器确认，此时retry参数不生效
  12.      * 1：leader写入记录到log，不会等待follower的确认即向生产者发送通知
  13.      * all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为-1
  14.      */privateString ack ="all";/**
  15.      * 生产者重试次数
  16.      */privateInteger retryTimes =1;/**
  17.      * 生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes=16K
  18.      */privateInteger batchSize =16384;/**
  19.      * 生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒
  20.      */privateInteger lingerMs =1;/**
  21.      * 生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes=32M.
  22.      */privateInteger bufferMemory =33554432;/**
  23.      * 生产者：键编码器
  24.      */privateString keyEncoder ="org.apache.kafka.common.serialization.StringSerializer";/**
  25.      * 生产者：值编码器
  26.      */privateString valueEncoder ="org.apache.kafka.common.serialization.StringSerializer";/**
  27.      * 消费者：消费topic的组ID
  28.      */privateString groupId ="my-group-id";/**
  29.      * 消费者：后台定期提交offset
  30.      */privateString autoCommit ="true";/**
  31.      * 消费者提交offset的时间间隔：单位：毫秒，当enable.auto.commit为true时生效
  32.      */privateString autoCommitIntervalMs ="1000";/**
  33.      * 消费者：键解码器
  34.      */privateString keyDecoder ="org.apache.kafka.common.serialization.StringDeserializer";/**
  35.      * 消费者：值解码器
  36.      */privateString valueDecoder ="org.apache.kafka.common.serialization.StringDeserializer";/**
  37.      * 消费者：重启后配置offset
  38.      * earliest：消费者恢复到当前topic最早的offset
  39.      * latest：消费者从最新的offset开始消费
  40.      * none：如果消费者组没找到之前的offset抛出异常
  41.      * 其他任何值都会抛出异常
  42.      */privateString autoOffsetReset ="latest";/**
  43.      * TOPIC
  44.      */privateCollection<String> topic =Collections.singleton("my-topic");publicKafkaCommonProperties(){}publicKafkaCommonProperties(String kafkaHost,String ack,Integer retryTimes,Integer batchSize,Integer lingerMs,Integer bufferMemory,String keyEncoder,String valueEncoder,String groupId,String autoCommit,String autoCommitIntervalMs,String keyDecoder,String valueDecoder,String autoOffsetReset,Collection<String> topic){this.kafkaHost = kafkaHost;this.ack = ack;this.retryTimes = retryTimes;this.batchSize = batchSize;this.lingerMs = lingerMs;this.bufferMemory = bufferMemory;this.keyEncoder = keyEncoder;this.valueEncoder = valueEncoder;this.groupId = groupId;this.autoCommit = autoCommit;this.autoCommitIntervalMs = autoCommitIntervalMs;this.keyDecoder = keyDecoder;this.valueDecoder = valueDecoder;this.autoOffsetReset = autoOffsetReset;this.topic = topic;}// 省略setter和getter及toString()  }

4 Code实践

4.1 生产者

构建Kafka数据生产者，
测试样例的配置有：Kafka broker地址，消息确认，重试，批量发送数据，数据键和值的编码器，
重写Callback实现异步生产数据。

4.1.1 生产数据

  1. packagecom.monkey.java_study.mq.kafka;importorg.apache.kafka.clients.producer.*;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importjava.util.Objects;importjava.util.Properties;/**
  2.  * Kafka生产者.
  3.  *
  4.  * @author xindaqi
  5.  * @since 2022-08-02 9:59
  6.  */publicclassKafkaProducerTest{privatestaticfinalLogger logger =LoggerFactory.getLogger(KafkaProducerTest.class);publicstaticKafkaProducer<String,String>getDefaultKafkaProducer(KafkaCommonProperties kafkaCommonProperties){Properties properties =newProperties();
  7.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getKafkaHost());
  8.         properties.put(ProducerConfig.ACKS_CONFIG, kafkaCommonProperties.getAck());
  9.         properties.put(ProducerConfig.RETRIES_CONFIG, kafkaCommonProperties.getRetryTimes());
  10.         properties.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaCommonProperties.getBatchSize());
  11.         properties.put(ProducerConfig.LINGER_MS_CONFIG, kafkaCommonProperties.getLingerMs());
  12.         properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaCommonProperties.getBufferMemory());
  13.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getKeyEncoder());
  14.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getValueEncoder());returnnewKafkaProducer<>(properties);}staticclassMyProducerCallbackimplementsCallback{@OverridepublicvoidonCompletion(RecordMetadata metadata,Exception exception){if(Objects.nonNull(exception)){
  15.                 logger.error(">>>>>>>>>>Producer生产消息异常：", exception);}if(Objects.nonNull(metadata)){
  16.                 logger.info(">>>>>>>>>>Producer生产消息：metadata：{}，partition:{}, offset：{}", metadata, metadata.partition(), metadata.offset());}}}publicstaticvoidmain(String[] args){KafkaCommonProperties kafkaCommonProperties =newKafkaCommonProperties();KafkaProducer<String,String> producer =getDefaultKafkaProducer(kafkaCommonProperties);String message ="hello world ";try{for(int i =0; i <10; i++){// 异步写入数据String topic = kafkaCommonProperties.getTopic().toArray()[0].toString();ProducerRecord<String,String> producerRecord =newProducerRecord<>(topic, message + i);
  17.                 producer.send(producerRecord,newMyProducerCallback());}}catch(Exception ex){
  18.             logger.error(">>>>>>>>生产数据异常：", ex);thrownewRuntimeException(ex);}finally{
  19.             producer.close();}}}

4.1.2 开启生产者

生产者开启后，控制台输出生产者配置信息，如下图所示，其中，
acks在代码中配置为all，而运行日志中acks为-1，所以，acks的all与-1是同种功能。
在这里插入图片描述
生产者生产数据是通过异步的方式，控制台日志如下图所示，
由图可知，生产数据的线程为：kafka-producer-network-thread。
在这里插入图片描述

4.2 消费者

Kafka消费者通过groupId消费指定topic的，
以groupId区分不同的消费者，即不同的groupId消费相同的topic，对于topic而言，就是不同的消费者，
同时，消费者需要记录消费到的offset，以便下次启动时定位到具体的位置，消费消息。
这里，配置的offset策略为：latest，即每次重启消费者时，从最新的offset开始消费（上次记录的offset之后的一个，如果上次消费没有记录，则从当前offset之后开始消费）。
offset的重置这样理解：
当前topic写入数据有4条，offset从0到3，
如果，offset重设为earliest，则每次重启消费者，offset都会从0开始消费数据；
如果，offset重设为latest，则，每次消费从上次消费的offset下一个开始消费，如果上次消费的offset为3，则，重启后， 从4开始消费数据。
在这里插入图片描述

4.2.1 消费数据

  1. packagecom.monkey.java_study.mq.kafka;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importjava.time.Duration;importjava.util.Properties;/**
  2.  * Kafka消费者.
  3.  *
  4.  * @author xindaqi
  5.  * @since 2022-08-02 9:59
  6.  */publicclassKafkaConsumerTest{privatestaticfinalLogger logger =LoggerFactory.getLogger(KafkaConsumerTest.class);publicstaticKafkaConsumer<String,String>getDefaultKafkaConsumer(KafkaCommonProperties kafkaCommonProperties){Properties properties =newProperties();
  7.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getKafkaHost());
  8.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaCommonProperties.getGroupId());
  9.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaCommonProperties.getAutoCommit());
  10.         properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaCommonProperties.getAutoCommitIntervalMs());
  11.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaCommonProperties.getAutoOffsetReset());
  12.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getKeyDecoder());
  13.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getValueDecoder());returnnewKafkaConsumer<>(properties);}publicstaticvoidmain(String[] args){try{KafkaCommonProperties kafkaCommonProperties =newKafkaCommonProperties();KafkaConsumer<String,String> consumer =getDefaultKafkaConsumer(kafkaCommonProperties);
  14.             consumer.subscribe(kafkaCommonProperties.getTopic());while(Boolean.TRUE){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String>record: records){
  15.                     logger.info(">>>>>>>>Consumer offset:{}, value:{}",record.offset(),record.value());}}}catch(Exception ex){thrownewRuntimeException(ex);}}}

4.2.2 开启消费者

开启消费者后，控制台输出消费者的配置参数，如下图所示。
在这里插入图片描述
同时输出的还有（如下图所示），下次将要消费的offset：41。
在这里插入图片描述
消费信息日志如下图所示，
由图可知，从offset 41开始消费。
在这里插入图片描述

5 命令行

上面

5.1 开启消费者

从topic的最开始消费数据，则offset从0开始，
开启消费者命令：

  1. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

topic中已有数据，因此，开始消费者时，从最开始消费数据，结果如下图所示。
在这里插入图片描述
命令行启动生产者后，
该消费者会自动消费进入数据，结果如下图所示。
在这里插入图片描述

5.2 开启生产者

开启生产者命令如下：

  1. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

发送消费如下图所示。
在这里插入图片描述

6 小结

无。

标签： kafka java
本文转载自: https://blog.csdn.net/Xin_101/article/details/126154171
版权归原作者 天然玩家 所有， 如有侵权，请联系我们删除。
登录后发布评论

“详细讲解如何使用Java连接Kafka构建生产者和消费者（带测试样例）”的评论:

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

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

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

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

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

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

Kafka学习笔记

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