文章目錄

• 說明
• Springboot集成Kafka
• • 依賴配置
  • 配置文件yml配置
  • Producer生產者
  • Consumer消費者
  • 測試代碼
• 注意事項

說明

本文是接《Kafka學習：CentOS7下Kafka集群搭建》后續，搭建完成kafka集群之后進行項目的集成。

Springboot集成Kafka

依賴配置

首先，在springboot中引入kafka的依賴jar包

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

配置文件yml配置

配置文件使用的是yml文件，其中配置如下所示：

# 端口，項目上下文 server:port: 8080servlet:context-path: /kafka-demo# kafka集群配置 spring:kafka:bootstrap-servers: hadoop-slave1:9092,hadoop-slave2:9092,hadoop-slave3:9092producer:# 發生錯誤后，消息重發的次數。retries: 0#當有多個消息需要被發送到同一個分區時，生產者會把它們放在同一個批次里。該參數指定了一個批次可以使用的內存大小，按照字節數計算。batch-size: 16384# 設置生產者內存緩沖區的大小。buffer-memory: 33554432# 鍵的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializer# 值的序列化方式value-serializer: org.apache.kafka.common.serialization.StringSerializer# acks=0 ： 生產者在成功寫入消息之前不會等待任何來自服務器的響應。# acks=1 ： 只要集群的首領節點收到消息，生產者就會收到一個來自服務器成功響應。# acks=all ：只有當所有參與復制的節點全部收到消息時，生產者才會收到一個來自服務器的成功響應。acks: 1consumer:# 設置kafka的消費者組group-id: mygroup1# 自動提交的時間間隔 在spring boot 2.X 版本中這里采用的是值的類型為Duration 需要符合特定的格式，如1S,1M,2H,5Dauto-commit-interval: 1S# 該屬性指定了消費者在讀取一個沒有偏移量的分區或者偏移量無效的情況下該作何處理：# latest（默認值）在偏移量無效的情況下，消費者將從最新的記錄開始讀取數據（在消費者啟動之后生成的記錄）# earliest ：在偏移量無效的情況下，消費者將從起始位置讀取分區的記錄auto-offset-reset: earliest# 是否自動提交偏移量，默認值是true,為了避免出現重復數據和數據丟失，可以把它設置為false,然后手動提交偏移量enable-auto-commit: false# 鍵的反序列化方式key-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 值的反序列化方式value-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:# 在偵聽器容器中運行的線程數。concurrency: 5#listner負責ack，每調用一次，就立即commitack-mode: manual_immediate# 如果Broker上不存在至少一個配置的主題（topic），則容器是否無法啟動，# 該設置項結合Broker設置項allow.auto.create.topics=true，如果為false，則會自動創建不存在的topicmissing-topics-fatal: false# 日志輸出配置 logging:level:root: INFOorg:springframework:security: WARNweb: ERRORfile:path: ./logsname: './logs/kafka-demo.log'pattern:file: '%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50}:%L - %msg%n'console: '%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50}:%L - %msg%n'

Producer生產者

生產者代碼如下所示：

package com.demo.kafka;import com.alibaba.fastjson.JSON; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback;/*** 模擬kafka生產者*/ @Component @Slf4j public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, Object> kafkaTemplate;//自定義topicpublic static final String TOPIC_TEST = "mytopic";//消費者組mygroup1public static final String TOPIC_GROUP1 = "mygroup1";//消費者組mygroup2public static final String TOPIC_GROUP2 = "mygroup2";public void send(Object obj) {String obj2String = JSON.toJSONString(obj);log.info("準備發送消息為：{}", obj2String);//發送消息ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TEST, obj);future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {@Overridepublic void onFailure(Throwable throwable) {//發送失敗的處理log.info(TOPIC_TEST + " - 生產者 發送消息失敗：" + throwable.getMessage());}@Overridepublic void onSuccess(SendResult<String, Object> stringObjectSendResult) {//成功的處理log.info(TOPIC_TEST + " - 生產者 發送消息成功：" + stringObjectSendResult.toString());}});} }

Consumer消費者

消費者代碼如下：

package com.demo.kafka;import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.handler.annotation.Header; import org.springframework.stereotype.Component;import java.util.Optional;/*** 模擬Consumer消費消息*/ @Component @Slf4j public class KafkaConsumer {@KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP1)public void topicTest(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {Optional<Object> message = Optional.ofNullable(record.value());if (message.isPresent()) {Object msg = message.get();log.info("topic_test 消費了： Topic:" + topic + ",Message:" + msg);ack.acknowledge();}}@KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP2)public void topicTest1(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {Optional<Object> message = Optional.ofNullable(record.value());if (message.isPresent()) {Object msg = message.get();log.info("topic_test1 消費了： Topic:" + topic + ",Message:" + msg);ack.acknowledge();}} }

測試代碼

使用一個簡單controller測試kafka發送和接收消息。

package com.demo.controller;import com.demo.kafka.KafkaProducer; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController;/*** 測試kafka發送和接收消息*/ @Slf4j @RestController @RequestMapping("/kafka") public class KafkaController {@Autowiredprivate KafkaProducer kafkaProducer;@GetMapping("/send")@Transactional(rollbackFor = Exception.class)public void send() {kafkaProducer.send("this is a test kafka topic message");} }

測試結果如下：

2021-01-09 22:39:50.341 [http-nio-8080-exec-1] INFO o.a.c.c.C.[Tomcat].[localhost].[/kafka-demo]:173 - Initializing Spring DispatcherServlet 'dispatcherServlet' 2021-01-09 22:39:50.395 [http-nio-8080-exec-1] INFO com.demo.kafka.KafkaProducer:30 - 準備發送消息為："this is a test kafka topic message" 2021-01-09 22:39:50.399 [http-nio-8080-exec-1] INFO org.apache.kafka.clients.producer.ProducerConfig:347 - ProducerConfig values: acks = 1batch.size = 16384bootstrap.servers = [hadoop-slave1:9092, hadoop-slave2:9092, hadoop-slave3:9092]buffer.memory = 33554432client.dns.lookup = defaultclient.id = producer-1compression.type = noneconnections.max.idle.ms = 540000delivery.timeout.ms = 120000enable.idempotence = falseinterceptor.classes = []key.serializer = class org.apache.kafka.common.serialization.StringSerializerlinger.ms = 0max.block.ms = 60000max.in.flight.requests.per.connection = 5max.request.size = 1048576metadata.max.age.ms = 300000metadata.max.idle.ms = 300000metric.reporters = []metrics.num.samples = 2metrics.recording.level = INFOmetrics.sample.window.ms = 30000partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitionerreceive.buffer.bytes = 32768reconnect.backoff.max.ms = 1000reconnect.backoff.ms = 50request.timeout.ms = 30000retries = 0retry.backoff.ms = 100sasl.client.callback.handler.class = nullsasl.jaas.config = nullsasl.kerberos.kinit.cmd = /usr/bin/kinitsasl.kerberos.min.time.before.relogin = 60000sasl.kerberos.service.name = nullsasl.kerberos.ticket.renew.jitter = 0.05sasl.kerberos.ticket.renew.window.factor = 0.8sasl.login.callback.handler.class = nullsasl.login.class = nullsasl.login.refresh.buffer.seconds = 300sasl.login.refresh.min.period.seconds = 60sasl.login.refresh.window.factor = 0.8sasl.login.refresh.window.jitter = 0.05sasl.mechanism = GSSAPIsecurity.protocol = PLAINTEXTsecurity.providers = nullsend.buffer.bytes = 131072ssl.cipher.suites = nullssl.enabled.protocols = [TLSv1.2]ssl.endpoint.identification.algorithm = httpsssl.key.password = nullssl.keymanager.algorithm = SunX509ssl.keystore.location = nullssl.keystore.password = nullssl.keystore.type = JKSssl.protocol = TLSv1.2ssl.provider = nullssl.secure.random.implementation = nullssl.trustmanager.algorithm = PKIXssl.truststore.location = nullssl.truststore.password = nullssl.truststore.type = JKStransaction.timeout.ms = 60000transactional.id = nullvalue.serializer = class org.apache.kafka.common.serialization.StringSerializer2021-01-09 22:39:50.409 [http-nio-8080-exec-1] INFO org.apache.kafka.common.utils.AppInfoParser:117 - Kafka version: 2.5.0 2021-01-09 22:39:50.410 [http-nio-8080-exec-1] INFO org.apache.kafka.common.utils.AppInfoParser:118 - Kafka commitId: 66563e712b0b9f84 2021-01-09 22:39:50.410 [http-nio-8080-exec-1] INFO org.apache.kafka.common.utils.AppInfoParser:119 - Kafka startTimeMs: 1610203190409 2021-01-09 22:39:50.418 [kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.Metadata:280 - [Producer clientId=producer-1] Cluster ID: DmchEHdzTi2ARIhv5RBBVQ 2021-01-09 22:39:50.443 [kafka-producer-network-thread | producer-1] INFO com.demo.kafka.KafkaProducer:43 - mytopic - 生產者 發送消息成功：SendResult [producerRecord=ProducerRecord(topic=mytopic, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=null, value=this is a test kafka topic message, timestamp=null), recordMetadata=mytopic-1@6] 2021-01-09 22:39:50.454 [org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1] INFO com.demo.kafka.KafkaConsumer:24 - topic_test 消費了： Topic:mytopic,Message:this is a test kafka topic message 2021-01-09 22:39:50.454 [org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1] INFO com.demo.kafka.KafkaConsumer:34 - topic_test1 消費了： Topic:mytopic,Message:this is a test kafka topic message

注意事項

需要注意的是，本文虛擬機kafka集群使用的都是域名，如果同樣是本地進行測試需要在host文件中添加域名和ip的映射。
hosts文件地址：

C:\Windows\System32\drivers\etc

例如，我這里添加的域名和ip的映射關系如下：

192.168.223.128 hadoop-slave1 192.168.223.129 hadoop-slave2 192.168.223.130 hadoop-slave3 192.168.223.131 hadoop-master

總結

以上是生活随笔為你收集整理的SpringBoot笔记：SpringBoot2.3集成Kafka组件配置的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。