SpringBoot集成kafka全面實戰(zhàn)記錄
本文是SpringBoot+Kafka的實戰(zhàn)講解，如果對kafka的架構(gòu)原理還不了解的讀者，建議先看一下《大白話kafka架構(gòu)原理》、《秒懂kafka HA（高可用）》兩篇文章。
一、生產(chǎn)者實踐
- 普通生產(chǎn)者
- 帶回調(diào)的生產(chǎn)者
- 自定義分區(qū)器
- kafka事務(wù)提交
二、消費者實踐
- 簡單消費
- 指定topic、partition、offset消費
- 批量消費
- 監(jiān)聽異常處理器
- 消息過濾器
- 消息轉(zhuǎn)發(fā)
- 定時啟動/停止監(jiān)聽器
一、前戲
1、在項目中連接kafka，因為是外網(wǎng)，首先要開放kafka配置文件中的如下配置（其中IP為公網(wǎng)IP），
advertised.listeners=PLAINTEXT://112.126.74.249:9092
2、在開始前我們先創(chuàng)建兩個topic：topic1、topic2，其分區(qū)和副本數(shù)都設(shè)置為
2，用來測試，
[root@iZ2zegzlkedbo3e64vkbefZ ~]# cd /usr/local/kafka-cluster/kafka1/bin/ [root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic1 Created topic topic1. [root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic2 Created topic topic2.
當(dāng)然我們也可以不手動創(chuàng)建topic，在執(zhí)行代碼kafkaTemplate.send("topic1", normalMessage)發(fā)送消息時，kafka會幫我們自動完成topic的創(chuàng)建工作，但這種情況下創(chuàng)建的topic默認(rèn)只有一個分區(qū)，分區(qū)也沒有副本。所以，我們可以在項目中新建一個配置類專門用來初始化topic，如下，
@Configuration
public class KafkaInitialConfiguration {
// 創(chuàng)建一個名為testtopic的Topic并設(shè)置分區(qū)數(shù)為8，分區(qū)副本數(shù)為2
@Bean
public NewTopic initialTopic() {
return new NewTopic("testtopic",8, (short) 2 );
}
?
// 如果要修改分區(qū)數(shù)，只需修改配置值重啟項目即可
// 修改分區(qū)數(shù)并不會導(dǎo)致數(shù)據(jù)的丟失，但是分區(qū)數(shù)只能增大不能減小
@Bean
public NewTopic updateTopic() {
return new NewTopic("testtopic",10, (short) 2 );
}
}
3、新建SpringBoot項目
① 引入pom依賴
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
② application.propertise配置（本文用到的配置項這里全列了出來）
###########【Kafka集群】########### spring.kafka.bootstrap-servers=112.126.74.249:9092,112.126.74.249:9093 ###########【初始化生產(chǎn)者配置】########### # 重試次數(shù) spring.kafka.producer.retries=0 # 應(yīng)答級別:多少個分區(qū)副本備份完成時向生產(chǎn)者發(fā)送ack確認(rèn)(可選0、1、all/-1) spring.kafka.producer.acks=1 # 批量大小 spring.kafka.producer.batch-size=16384 # 提交延時 spring.kafka.producer.properties.linger.ms=0 # 當(dāng)生產(chǎn)端積累的消息達(dá)到batch-size或接收到消息linger.ms后,生產(chǎn)者就會將消息提交給kafka # linger.ms為0表示每接收到一條消息就提交給kafka,這時候batch-size其實就沒用了 ? # 生產(chǎn)端緩沖區(qū)大小 spring.kafka.producer.buffer-memory = 33554432 # Kafka提供的序列化和反序列化類 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer # 自定義分區(qū)器 # spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner ? ###########【初始化消費者配置】########### # 默認(rèn)的消費組ID spring.kafka.consumer.properties.group.id=defaultConsumerGroup # 是否自動提交offset spring.kafka.consumer.enable-auto-commit=true # 提交offset延時(接收到消息后多久提交offset) spring.kafka.consumer.auto.commit.interval.ms=1000 # 當(dāng)kafka中沒有初始o(jì)ffset或offset超出范圍時將自動重置offset # earliest:重置為分區(qū)中最小的offset; # latest:重置為分區(qū)中最新的offset(消費分區(qū)中新產(chǎn)生的數(shù)據(jù)); # none:只要有一個分區(qū)不存在已提交的offset,就拋出異常; spring.kafka.consumer.auto-offset-reset=latest # 消費會話超時時間(超過這個時間consumer沒有發(fā)送心跳,就會觸發(fā)rebalance操作) spring.kafka.consumer.properties.session.timeout.ms=120000 # 消費請求超時時間 spring.kafka.consumer.properties.request.timeout.ms=180000 # Kafka提供的序列化和反序列化類 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer # 消費端監(jiān)聽的topic不存在時，項目啟動會報錯(關(guān)掉) spring.kafka.listener.missing-topics-fatal=false # 設(shè)置批量消費 # spring.kafka.listener.type=batch # 批量消費每次最多消費多少條消息 # spring.kafka.consumer.max-poll-records=50
二、Hello Kafka
1、簡單生產(chǎn)者
@RestController
public class KafkaProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
?
// 發(fā)送消息
@GetMapping("/kafka/normal/{message}")
public void sendMessage1(@PathVariable("message") String normalMessage) {
kafkaTemplate.send("topic1", normalMessage);
}
}
?2、簡單消費
@Component
public class KafkaConsumer {
// 消費監(jiān)聽
@KafkaListener(topics = {"topic1"})
public void onMessage1(ConsumerRecord<?, ?> record){
// 消費的哪個topic、partition的消息,打印出消息內(nèi)容
System.out.println("簡單消費："+record.topic()+"-"+record.partition()+"-"+record.value());
}
}
上面示例創(chuàng)建了一個生產(chǎn)者，發(fā)送消息到topic1，消費者監(jiān)聽topic1消費消息。監(jiān)聽器用@KafkaListener注解，topics表示監(jiān)聽的topic，支持同時監(jiān)聽多個，用英文逗號分隔。啟動項目，postman調(diào)接口觸發(fā)生產(chǎn)者發(fā)送消息，
可以看到監(jiān)聽器消費成功，
三、生產(chǎn)者
1、帶回調(diào)的生產(chǎn)者
kafkaTemplate提供了一個回調(diào)方法addCallback，我們可以在回調(diào)方法中監(jiān)控消息是否發(fā)送成功 或 失敗時做補(bǔ)償處理，有兩種寫法，
@GetMapping("/kafka/callbackOne/{message}")
public void sendMessage2(@PathVariable("message") String callbackMessage) {
kafkaTemplate.send("topic1", callbackMessage).addCallback(success -> {
// 消息發(fā)送到的topic
String topic = success.getRecordMetadata().topic();
// 消息發(fā)送到的分區(qū)
int partition = success.getRecordMetadata().partition();
// 消息在分區(qū)內(nèi)的offset
long offset = success.getRecordMetadata().offset();
System.out.println("發(fā)送消息成功:" + topic + "-" + partition + "-" + offset);
}, failure -> {
System.out.println("發(fā)送消息失敗:" + failure.getMessage());
});
}
@GetMapping("/kafka/callbackTwo/{message}")
public void sendMessage3(@PathVariable("message") String callbackMessage) {
kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) {
System.out.println("發(fā)送消息失?。?+ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
System.out.println("發(fā)送消息成功：" + result.getRecordMetadata().topic() + "-"
+ result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
}
});
}
2、自定義分區(qū)器
我們知道，kafka中每個topic被劃分為多個分區(qū)，那么生產(chǎn)者將消息發(fā)送到topic時，具體追加到哪個分區(qū)呢？這就是所謂的分區(qū)策略，Kafka 為我們提供了默認(rèn)的分區(qū)策略，同時它也支持自定義分區(qū)策略。其路由機(jī)制為：
① 若發(fā)送消息時指定了分區(qū)（即自定義分區(qū)策略），則直接將消息append到指定分區(qū)；
② 若發(fā)送消息時未指定 patition，但指定了 key（kafka允許為每條消息設(shè)置一個key），則對key值進(jìn)行hash計算，根據(jù)計算結(jié)果路由到指定分區(qū)，這種情況下可以保證同一個 Key 的所有消息都進(jìn)入到相同的分區(qū)；
③ ?patition 和 key 都未指定，則使用kafka默認(rèn)的分區(qū)策略，輪詢選出一個 patition；
※ 我們來自定義一個分區(qū)策略，將消息發(fā)送到我們指定的partition，首先新建一個分區(qū)器類實現(xiàn)Partitioner接口，重寫方法，其中partition方法的返回值就表示將消息發(fā)送到幾號分區(qū)，
public class CustomizePartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 自定義分區(qū)規(guī)則(這里假設(shè)全部發(fā)到0號分區(qū))
// ......
return 0;
}
?
@Override
public void close() {
?
}
?
@Override
public void configure(Map<String, ?> configs) {
?
}
}
在application.propertise中配置自定義分區(qū)器，配置的值就是分區(qū)器類的全路徑名，
# 自定義分區(qū)器 spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
3、kafka事務(wù)提交
如果在發(fā)送消息時需要創(chuàng)建事務(wù)，可以使用 KafkaTemplate 的 executeInTransaction 方法來聲明事務(wù)，
@GetMapping("/kafka/transaction")
public void sendMessage7(){
// 聲明事務(wù)：后面報錯消息不會發(fā)出去
kafkaTemplate.executeInTransaction(operations -> {
operations.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
});
?
// 不聲明事務(wù)：后面報錯但前面消息已經(jīng)發(fā)送成功了
kafkaTemplate.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
}
四、消費者
1、指定topic、partition、offset消費
前面我們在監(jiān)聽消費topic1的時候，監(jiān)聽的是topic1上所有的消息，如果我們想指定topic、指定partition、指定offset來消費呢？也很簡單，@KafkaListener注解已全部為我們提供，
# 設(shè)置批量消費 spring.kafka.listener.type=batch # 批量消費每次最多消費多少條消息 spring.kafka.consumer.max-poll-records=50
屬性解釋：
① id：消費者ID；
② groupId：消費組ID；
③ topics：監(jiān)聽的topic，可監(jiān)聽多個；
④ topicPartitions：可配置更加詳細(xì)的監(jiān)聽信息，可指定topic、parition、offset監(jiān)聽。
上面onMessage2監(jiān)聽的含義：監(jiān)聽topic1的0號分區(qū)，同時監(jiān)聽topic2的0號分區(qū)和topic2的1號分區(qū)里面offset從8開始的消息。
注意：topics和topicPartitions不能同時使用；
2、批量消費
設(shè)置application.prpertise開啟批量消費即可，
@KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic1")
public void onMessage3(List<ConsumerRecord<?, ?>> records) {
System.out.println(">>>批量消費一次，records.size()="+records.size());
for (ConsumerRecord<?, ?> record : records) {
System.out.println(record.value());
}
}
接收消息時用List來接收，監(jiān)聽代碼如下，
// 新建一個異常處理器，用@Bean注入
@Bean
public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
return (message, exception, consumer) -> {
System.out.println("消費異常："+message.getPayload());
return null;
};
}
?
// 將這個異常處理器的BeanName放到@KafkaListener注解的errorHandler屬性里面
@KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
throw new Exception("簡單消費-模擬異常");
}
?
// 批量消費也一樣，異常處理器的message.getPayload()也可以拿到各條消息的信息
@KafkaListener(topics = "topic1",errorHandler="consumerAwareErrorHandler")
public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
System.out.println("批量消費一次...");
throw new Exception("批量消費-模擬異常");
}
3、ConsumerAwareListenerErrorHandler 異常處理器
通過異常處理器，我們可以處理consumer在消費時發(fā)生的異常。
新建一個?ConsumerAwareListenerErrorHandler 類型的異常處理方法，用@Bean注入，BeanName默認(rèn)就是方法名，然后我們將這個異常處理器的BeanName放到@KafkaListener注解的errorHandler屬性里面，當(dāng)監(jiān)聽拋出異常的時候，則會自動調(diào)用異常處理器，
@Component
public class KafkaConsumer {
@Autowired
ConsumerFactory consumerFactory;
?
// 消息過濾器
@Bean
public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory);
// 被過濾的消息將被丟棄
factory.setAckDiscarded(true);
// 消息過濾策略
factory.setRecordFilterStrategy(consumerRecord -> {
if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
return false;
}
//返回true消息則被過濾
return true;
});
return factory;
}
?
// 消息過濾監(jiān)聽
@KafkaListener(topics = {"topic1"},containerFactory = "filterContainerFactory")
public void onMessage6(ConsumerRecord<?, ?> record) {
System.out.println(record.value());
}
}
執(zhí)行看一下效果，
4、消息過濾器
消息過濾器可以在消息抵達(dá)consumer之前被攔截，在實際應(yīng)用中，我們可以根據(jù)自己的業(yè)務(wù)邏輯，篩選出需要的信息再交由KafkaListener處理，不需要的消息則過濾掉。
配置消息過濾只需要為 監(jiān)聽器工廠 配置一個RecordFilterStrategy（消息過濾策略），返回true的時候消息將會被拋棄，返回false時，消息能正常抵達(dá)監(jiān)聽容器。
/**
* @Title 消息轉(zhuǎn)發(fā)
* @Description 從topic1接收到的消息經(jīng)過處理后轉(zhuǎn)發(fā)到topic2
* @Author long.yuan
* @Date 2020/3/23 22:15
* @Param [record]
* @return void
**/
@KafkaListener(topics = {"topic1"})
@SendTo("topic2")
public String onMessage7(ConsumerRecord<?, ?> record) {
return record.value()+"-forward message";
}
上面實現(xiàn)了一個"過濾奇數(shù)、接收偶數(shù)"的過濾策略，我們向topic1發(fā)送0-99總共100條消息，看一下監(jiān)聽器的消費情況，可以看到監(jiān)聽器只消費了偶數(shù)，
5、消息轉(zhuǎn)發(fā)
在實際開發(fā)中，我們可能有這樣的需求，應(yīng)用A從TopicA獲取到消息，經(jīng)過處理后轉(zhuǎn)發(fā)到TopicB，再由應(yīng)用B監(jiān)聽處理消息，即一個應(yīng)用處理完成后將該消息轉(zhuǎn)發(fā)至其他應(yīng)用，完成消息的轉(zhuǎn)發(fā)。
在SpringBoot集成Kafka實現(xiàn)消息的轉(zhuǎn)發(fā)也很簡單，只需要通過一個@SendTo注解，被注解方法的return值即轉(zhuǎn)發(fā)的消息內(nèi)容，如下，
/**
* @Title 消息轉(zhuǎn)發(fā)
* @Description 從topic1接收到的消息經(jīng)過處理后轉(zhuǎn)發(fā)到topic2
* @Author long.yuan
* @Date 2020/3/23 22:15
* @Param [record]
* @return void
**/
@KafkaListener(topics = {"topic1"})
@SendTo("topic2")
public String onMessage7(ConsumerRecord<?, ?> record) {
return record.value()+"-forward message";
}
6、定時啟動、停止監(jiān)聽器
默認(rèn)情況下，當(dāng)消費者項目啟動的時候，監(jiān)聽器就開始工作，監(jiān)聽消費發(fā)送到指定topic的消息，那如果我們不想讓監(jiān)聽器立即工作，想讓它在我們指定的時間點開始工作，或者在我們指定的時間點停止工作，該怎么處理呢——使用KafkaListenerEndpointRegistry，下面我們就來實現(xiàn)：
① 禁止監(jiān)聽器自啟動；
② 創(chuàng)建兩個定時任務(wù)，一個用來在指定時間點啟動定時器，另一個在指定時間點停止定時器；
新建一個定時任務(wù)類，用注解@EnableScheduling聲明，KafkaListenerEndpointRegistry 在SpringIO中已經(jīng)被注冊為Bean，直接注入，設(shè)置禁止KafkaListener自啟動，
@EnableScheduling
@Component
public class CronTimer {
?
/**
* @KafkaListener注解所標(biāo)注的方法并不會在IOC容器中被注冊為Bean，
* 而是會被注冊在KafkaListenerEndpointRegistry中，
* 而KafkaListenerEndpointRegistry在SpringIOC中已經(jīng)被注冊為Bean
**/
@Autowired
private KafkaListenerEndpointRegistry registry;
?
@Autowired
private ConsumerFactory consumerFactory;
?
// 監(jiān)聽器容器工廠(設(shè)置禁止KafkaListener自啟動)
@Bean
public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
container.setConsumerFactory(consumerFactory);
//禁止KafkaListener自啟動
container.setAutoStartup(false);
return container;
}
?
// 監(jiān)聽器
@KafkaListener(id="timingConsumer",topics = "topic1",containerFactory = "delayContainerFactory")
public void onMessage1(ConsumerRecord<?, ?> record){
System.out.println("消費成功："+record.topic()+"-"+record.partition()+"-"+record.value());
}
?
// 定時啟動監(jiān)聽器
@Scheduled(cron = "0 42 11 * * ? ")
public void startListener() {
System.out.println("啟動監(jiān)聽器...");
// "timingConsumer"是@KafkaListener注解后面設(shè)置的監(jiān)聽器ID,標(biāo)識這個監(jiān)聽器
if (!registry.getListenerContainer("timingConsumer").isRunning()) {
registry.getListenerContainer("timingConsumer").start();
}
//registry.getListenerContainer("timingConsumer").resume();
}
?
// 定時停止監(jiān)聽器
@Scheduled(cron = "0 45 11 * * ? ")
public void shutDownListener() {
System.out.println("關(guān)閉監(jiān)聽器...");
registry.getListenerContainer("timingConsumer").pause();
}
}
啟動項目，觸發(fā)生產(chǎn)者向topic1發(fā)送消息，可以看到consumer沒有消費，因為這時監(jiān)聽器還沒有開始工作，
11:42分監(jiān)聽器啟動開始工作，消費消息，
11：45分監(jiān)聽器停止工作，
到此這篇關(guān)于SpringBoot集成kafka全面實戰(zhàn)的文章就介紹到這了,更多相關(guān)SpringBoot集成kafka內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Java 反射機(jī)制知識詳細(xì)介紹及總結(jié)
反射機(jī)制是在運行狀態(tài)中，對于任意一個類，都能夠知道這個類的所有屬性和方法；對于任意一個對象，都能夠調(diào)用它的任意一個方法和屬性；這種動態(tài)獲取的信息以及動態(tài)調(diào)用對象的方法的功能稱為java語言的反射機(jī)制2017-01-01
SpringBoot多環(huán)境配置及配置文件分類實例詳解
這篇文章主要介紹了SpringBoot多環(huán)境配置及配置文件分類,本文通過實例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-10-10
一個運行中的程序,?產(chǎn)生的對象是大量的,?如果對象不被繼續(xù)使用,?就會成為垃圾,?最后越堆越多,?最后占滿內(nèi)存,?所以我們要對這些垃圾進(jìn)行回收,保持程序的正常運行2022-05-05
Java 關(guān)于時間復(fù)雜度和空間復(fù)雜度的深度刨析
算法復(fù)雜度分為時間復(fù)雜度和空間復(fù)雜度。其作用： 時間復(fù)雜度是度量算法執(zhí)行的時間長短；而空間復(fù)雜度是度量算法所需存儲空間的大小2021-11-11
Java數(shù)據(jù)結(jié)構(gòu)之哈夫曼樹概述及實現(xiàn)
文中詳細(xì)講了關(guān)于Java哈夫曼樹的概述以及用Java實現(xiàn)的方法,對各位正在學(xué)習(xí)java數(shù)據(jù)結(jié)構(gòu)的小伙伴們有很大的幫助喲,需要的朋友可以參考下2021-05-05
這篇文章主要介紹了java鎖synchronized面試常問總結(jié)分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-12-12
String.join() 方法是連接指定數(shù)組的元素或集合的成員,在每個元素或成員之間使用指定的分隔符,這篇文章主要介紹了String.join()方法示例詳解,需要的朋友可以參考下2024-01-01
區(qū)分java中String+String和String+char
這篇文章主要向大家詳細(xì)區(qū)分了java中String+String和String+char，感興趣的小伙伴們可以參考一下2016-01-01
在IDEA中創(chuàng)建跑得起來的Springboot項目
這篇文章主要介紹了在IDEA中創(chuàng)建跑得起來的Springboot項目的圖文教程,需要的朋友可以參考下2018-04-04
SpringBoot中使用Servlet的兩種方式小結(jié)
這篇文章主要介紹了SpringBoot中使用Servlet的兩種方式小結(jié)，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-07-07