详细讲解如何使用Java连接Kafka构建生产者和消费者

　　前言
　　学习消息队列的过程中，先补习了RabbitMQ相关知识，接着又重温了Kafka相关的知识，发现，我并没有积累Java原生操作Kafka的文章，只使用SpringBoot集成过Kafka，所以，本次是纯Java的方式操作Kafka，构建生产者和消费者，本地部署Kafka环境，给出测试样例的测试结果，同时，讲解部分通用的参数，及给出通过命令行启动生产者和消费者的测试样例，分享如下，帮助读者学习Kafka基础操作。2环境准备
　　下载kafka2。1启动zookeeperbinzookeeperserverstart。shconfigzookeeper。properties2。2启动kafkabinkafkaserverstart。shconfigserver。properties2。3新建topicbinkafkatopics。shcreatezookeeperlocalhost：2181replicationfactor1partitions1topicmytopic
　　2。4依赖dependencygroupIdorg。apache。kafkagroupIdkafkaclientsartifactIdversion3。2。0versiondependency3Kafka通用配置
　　Kafka的生产者和消费者参数比较多，这里仅列出一些测试用的参数，
　　区分生产者和消费者。3。1生产者配置参数
　　序号
　　参数
　　描述
　　1hrbootstrap。servers
　　Kafka主机
　　2hracks
　　生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化
　　消息确认：
　　0：生产者不等待服务器确认，此时retry参数不生效
　　1：leader写入记录到log，不会等待follower的确认即向生产者发送通知
　　all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为1
　　3hrretries
　　生产者重试次数
　　4hrbatch。size
　　生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes16K
　　5hrlinger。ms
　　生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒
　　6hrbuffer。memory
　　生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes32M
　　7hrkey。serializer
　　生产者：键编码器
　　8hrvalue。serializer
　　生产者：值编码器3。2消费者配置参数
　　序号
　　参数
　　描述
　　1hrbootstrap。servers
　　Kafka主机
　　2hrgroup。id
　　消费者：消费topic的组ID
　　3hrenable。auto。commit
　　消费者：后台定期提交offset
　　4hrauto。commit。interval。ms
　　消费者提交offset的时间间隔：单位：毫秒，当enable。auto。commit为true时生效
　　5hrauto。offset。reset
　　消费者：重启后配置offset
　　earliest：消费者恢复到当前topic最早的offset
　　latest：消费者从最新的offset开始消费
　　none：如果消费者组没找到之前的offset抛出异常
　　其他任何值都会抛出异常
　　6hrkey。deserializer
　　消费者：键解码器
　　7hrvalue。deserializer
　　消费者：值解码器3。3Kafka通用参数封装
　　由于参数众多，这里封装了一个Kafka通用参数类，给了默认值，
　　本地测试，直接使用默认参数，
　　同时给出了有参构造器，自定义参数，
　　代码样例如下。packagecom。monkey。javastudy。mq。importjava。util。Cimportjava。util。CKafka通用配置。authorxindaqisince202208039：49publicclassKafkaCommonProperties｛Kafka主机privateStringkafkaHost192。168。211。129：9092；生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化消息确认：0：生产者不等待服务器确认，此时retry参数不生效1：leader写入记录到log，不会等待follower的确认即向生产者发送通知all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为1privateS生产者重试次数privateIntegerretryTimes1；生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes16KprivateIntegerbatchSize16384；生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒privateIntegerlingerMs1；生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes32M。privateIntegerbufferMemory33554432；生产者：键编码器privateStringkeyEncoderorg。apache。kafka。common。serialization。StringS生产者：值编码器privateStringvalueEncoderorg。apache。kafka。common。serialization。StringS消费者：消费topic的组IDprivateStringgroupI消费者：后台定期提交offsetprivateStringautoC消费者提交offset的时间间隔：单位：毫秒，当enable。auto。commit为true时生效privateStringautoCommitIntervalMs1000；消费者：键解码器privateStringkeyDecoderorg。apache。kafka。common。serialization。StringD消费者：值解码器privateStringvalueDecoderorg。apache。kafka。common。serialization。StringD消费者：重启后配置offsetearliest：消费者恢复到当前topic最早的offsetlatest：消费者从最新的offset开始消费none：如果消费者组没找到之前的offset抛出异常其他任何值都会抛出异常privateStringautoOffsetRTOPICprivateCollectionStringtopicCollections。singleton（mytopic）；publicKafkaCommonProperties（）｛｝publicKafkaCommonProperties（StringkafkaHost，Stringack，IntegerretryTimes，IntegerbatchSize，IntegerlingerMs，IntegerbufferMemory，StringkeyEncoder，StringvalueEncoder，StringgroupId，StringautoCommit，StringautoCommitIntervalMs，StringkeyDecoder，StringvalueDecoder，StringautoOffsetReset，CollectionStringtopic）｛this。kafkaHostkafkaHthis。this。retryTimesretryTthis。batchSizebatchSthis。lingerMslingerMs；this。bufferMemorybufferMthis。keyEncoderkeyEthis。valueEncodervalueEthis。groupIdgroupId；this。autoCommitautoCthis。autoCommitIntervalMsautoCommitIntervalMs；this。keyDecoderkeyDthis。valueDecodervalueDthis。autoOffsetResetautoOffsetRthis。｝省略setter和getter及toString（）｝4Code实践4。1生产者
　　构建Kafka数据生产者，
　　测试样例的配置有：Kafkabroker地址，消息确认，重试，批量发送数据，数据键和值的编码器，
　　重写Callback实现异步生产数据。4。1。1生产数据packagecom。monkey。javastudy。mq。importorg。apache。kafka。clients。producer。；importorg。slf4j。Limportorg。slf4j。LoggerFimportjava。util。Oimportjava。util。PKafka生产者。authorxindaqisince202208029：59publicclassKafkaProducerTest｛privatestaticfinalLoggerloggerLoggerFactory。getLogger（KafkaProducerTest。class）；publicstaticKafkaProducerString，StringgetDefaultKafkaProducer（KafkaCommonPropertieskafkaCommonProperties）｛PropertiespropertiesnewProperties（）；properties。put（ProducerConfig。BOOTSTRAPSERVERSCONFIG，kafkaCommonProperties。getKafkaHost（））；properties。put（ProducerConfig。ACKSCONFIG，kafkaCommonProperties。getAck（））；properties。put（ProducerConfig。RETRIESCONFIG，kafkaCommonProperties。getRetryTimes（））；properties。put（ProducerConfig。BATCHSIZECONFIG，kafkaCommonProperties。getBatchSize（））；properties。put（ProducerConfig。LINGERMSCONFIG，kafkaCommonProperties。getLingerMs（））；properties。put（ProducerConfig。BUFFERMEMORYCONFIG，kafkaCommonProperties。getBufferMemory（））；properties。put（ProducerConfig。KEYSERIALIZERCLASSCONFIG，kafkaCommonProperties。getKeyEncoder（））；properties。put（ProducerConfig。VALUESERIALIZERCLASSCONFIG，kafkaCommonProperties。getValueEncoder（））；returnnewKafkaProducer（properties）；｝staticclassMyProducerCallbackimplementsCallback｛OverridepublicvoidonCompletion（RecordMetadatametadata，Exceptionexception）｛if（Objects。nonNull（exception））｛logger。error（Producer生产消息异常：，exception）；｝if（Objects。nonNull（metadata））｛logger。info（Producer生产消息：metadata：｛｝，partition：｛｝，offset：｛｝，metadata，metadata。partition（），metadata。offset（））；｝｝｝publicstaticvoidmain（String〔〕args）｛KafkaCommonPropertieskafkaCommonPropertiesnewKafkaCommonProperties（）；KafkaProducerString，StringproducergetDefaultKafkaProducer（kafkaCommonProperties）；Stry｛for（inti0；i10；i）｛异步写入数据StringtopickafkaCommonProperties。getTopic（）。toArray（）〔0〕。toString（）；ProducerRecordString，StringproducerRecordnewProducerRecord（topic，messagei）；producer。send（producerRecord，newMyProducerCallback（））；｝｝catch（Exceptionex）｛logger。error（生产数据异常：，ex）；thrownewRuntimeException（ex）；｝finally｛producer。close（）；｝｝｝4。1。2开启生产者
　　生产者开启后，控制台输出生产者配置信息，如下图所示，其中，
　　acks在代码中配置为all，而运行日志中acks为1，所以，acks的all与1是同种功能。
　　生产者生产数据是通过异步的方式，控制台日志如下图所示，
　　由图可知，生产数据的线程为：kafkaproducernetworkthread。
　　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消费数据packagecom。monkey。javastudy。mq。importorg。apache。kafka。clients。consumer。ConsumerCimportorg。apache。kafka。clients。consumer。ConsumerRimportorg。apache。kafka。clients。consumer。ConsumerRimportorg。apache。kafka。clients。consumer。KafkaCimportorg。slf4j。Limportorg。slf4j。LoggerFimportjava。time。Dimportjava。util。PKafka消费者。authorxindaqisince202208029：59publicclassKafkaConsumerTest｛privatestaticfinalLoggerloggerLoggerFactory。getLogger（KafkaConsumerTest。class）；publicstaticKafkaConsumerString，StringgetDefaultKafkaConsumer（KafkaCommonPropertieskafkaCommonProperties）｛PropertiespropertiesnewProperties（）；properties。put（ConsumerConfig。BOOTSTRAPSERVERSCONFIG，kafkaCommonProperties。getKafkaHost（））；properties。put（ConsumerConfig。GROUPIDCONFIG，kafkaCommonProperties。getGroupId（））；properties。put（ConsumerConfig。ENABLEAUTOCOMMITCONFIG，kafkaCommonProperties。getAutoCommit（））；properties。put（ConsumerConfig。AUTOCOMMITINTERVALMSCONFIG，kafkaCommonProperties。getAutoCommitIntervalMs（））；properties。put（ConsumerConfig。AUTOOFFSETRESETCONFIG，kafkaCommonProperties。getAutoOffsetReset（））；properties。put（ConsumerConfig。KEYDESERIALIZERCLASSCONFIG，kafkaCommonProperties。getKeyDecoder（））；properties。put（ConsumerConfig。VALUEDESERIALIZERCLASSCONFIG，kafkaCommonProperties。getValueDecoder（））；returnnewKafkaConsumer（properties）；｝publicstaticvoidmain（String〔〕args）｛try｛KafkaCommonPropertieskafkaCommonPropertiesnewKafkaCommonProperties（）；KafkaConsumerString，StringconsumergetDefaultKafkaConsumer（kafkaCommonProperties）；consumer。subscribe（kafkaCommonProperties。getTopic（））；while（Boolean。TRUE）｛ConsumerRecordsString，Stringrecordsconsumer。poll（Duration。ofMillis（1000））；for（ConsumerRecordString，Stringrecord：records）｛logger。info（Consumeroffset：｛｝，value：｛｝，record。offset（），record。value（））；｝｝｝catch（Exceptionex）｛thrownewRuntimeException（ex）；｝｝｝4。2。2开启消费者
　　开启消费者后，控制台输出消费者的配置参数，如下图所示。
　　同时输出的还有（如下图所示），下次将要消费的offset：41。
　　消费信息日志如下图所示，
　　由图可知，从offset41开始消费。
　　5命令行
　　上面5。1开启消费者
　　从topic的最开始消费数据，则offset从0开始，
　　开启消费者命令：binkafkaconsoleconsumer。shbootstrapserverlocalhost：9092topicmytopicfrombeginning
　　topic中已有数据，因此，开始消费者时，从最开始消费数据，结果如下图所示。
　　命令行启动生产者后，
　　该消费者会自动消费进入数据，结果如下图所示。
　　5。2开启生产者
　　开启生产者命令如下：binkafkaconsoleproducer。shbrokerlistlocalhost：9092topicmytopic
　　发送消费如下图所示。
　　无。