SpringBoot集成kafka全面實戰(zhàn)記錄
本文是SpringBoot+Kafka的實戰(zhàn)講解，如果對kafka的架構原理還不了解的讀者，建議先看一下《大白話kafka架構原理》、《秒懂kafka HA（高可用）》兩篇文章。
一、生產者實踐
- 普通生產者
- 帶回調的生產者
- 自定義分區(qū)器
- kafka事務提交
二、消費者實踐
- 簡單消費
- 指定topic、partition、offset消費
- 批量消費
- 監(jiān)聽異常處理器
- 消息過濾器
- 消息轉發(fā)
- 定時啟動/停止監(jiān)聽器
一、前戲
1、在項目中連接kafka，因為是外網，首先要開放kafka配置文件中的如下配置（其中IP為公網IP），
advertised.listeners=PLAINTEXT://112.126.74.249:9092
2、在開始前我們先創(chuàng)建兩個topic：topic1、topic2，其分區(qū)和副本數都設置為
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.
當然我們也可以不手動創(chuàng)建topic，在執(zhí)行代碼kafkaTemplate.send("topic1", normalMessage)發(fā)送消息時，kafka會幫我們自動完成topic的創(chuàng)建工作，但這種情況下創(chuàng)建的topic默認只有一個分區(qū)，分區(qū)也沒有副本。所以，我們可以在項目中新建一個配置類專門用來初始化topic，如下，
@Configuration
public class KafkaInitialConfiguration {
// 創(chuàng)建一個名為testtopic的Topic并設置分區(qū)數為8，分區(qū)副本數為2
@Bean
public NewTopic initialTopic() {
return new NewTopic("testtopic",8, (short) 2 );
}
?
// 如果要修改分區(qū)數，只需修改配置值重啟項目即可
// 修改分區(qū)數并不會導致數據的丟失，但是分區(qū)數只能增大不能減小
@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 ###########【初始化生產者配置】########### # 重試次數 spring.kafka.producer.retries=0 # 應答級別:多少個分區(qū)副本備份完成時向生產者發(fā)送ack確認(可選0、1、all/-1) spring.kafka.producer.acks=1 # 批量大小 spring.kafka.producer.batch-size=16384 # 提交延時 spring.kafka.producer.properties.linger.ms=0 # 當生產端積累的消息達到batch-size或接收到消息linger.ms后,生產者就會將消息提交給kafka # linger.ms為0表示每接收到一條消息就提交給kafka,這時候batch-size其實就沒用了 ? # 生產端緩沖區(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 ? ###########【初始化消費者配置】########### # 默認的消費組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 # 當kafka中沒有初始offset或offset超出范圍時將自動重置offset # earliest:重置為分區(qū)中最小的offset; # latest:重置為分區(qū)中最新的offset(消費分區(qū)中新產生的數據); # 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不存在時，項目啟動會報錯(關掉) spring.kafka.listener.missing-topics-fatal=false # 設置批量消費 # spring.kafka.listener.type=batch # 批量消費每次最多消費多少條消息 # spring.kafka.consumer.max-poll-records=50
二、Hello Kafka
1、簡單生產者
@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的消息,打印出消息內容
System.out.println("簡單消費："+record.topic()+"-"+record.partition()+"-"+record.value());
}
}
上面示例創(chuàng)建了一個生產者，發(fā)送消息到topic1，消費者監(jiān)聽topic1消費消息。監(jiān)聽器用@KafkaListener注解，topics表示監(jiān)聽的topic，支持同時監(jiān)聽多個，用英文逗號分隔。啟動項目，postman調接口觸發(fā)生產者發(fā)送消息，
可以看到監(jiān)聽器消費成功，
三、生產者
1、帶回調的生產者
kafkaTemplate提供了一個回調方法addCallback，我們可以在回調方法中監(jiān)控消息是否發(fā)送成功 或 失敗時做補償處理，有兩種寫法，
@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ū)內的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ū)，那么生產者將消息發(fā)送到topic時，具體追加到哪個分區(qū)呢？這就是所謂的分區(qū)策略，Kafka 為我們提供了默認的分區(qū)策略，同時它也支持自定義分區(qū)策略。其路由機制為：
① 若發(fā)送消息時指定了分區(qū)（即自定義分區(qū)策略），則直接將消息append到指定分區(qū)；
② 若發(fā)送消息時未指定 patition，但指定了 key（kafka允許為每條消息設置一個key），則對key值進行hash計算，根據計算結果路由到指定分區(qū)，這種情況下可以保證同一個 Key 的所有消息都進入到相同的分區(qū)；
③ ?patition 和 key 都未指定，則使用kafka默認的分區(qū)策略，輪詢選出一個 patition；
※ 我們來自定義一個分區(qū)策略，將消息發(fā)送到我們指定的partition，首先新建一個分區(qū)器類實現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ī)則(這里假設全部發(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事務提交
如果在發(fā)送消息時需要創(chuàng)建事務，可以使用 KafkaTemplate 的 executeInTransaction 方法來聲明事務，
@GetMapping("/kafka/transaction")
public void sendMessage7(){
// 聲明事務：后面報錯消息不會發(fā)出去
kafkaTemplate.executeInTransaction(operations -> {
operations.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
});
?
// 不聲明事務：后面報錯但前面消息已經發(fā)送成功了
kafkaTemplate.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
}
四、消費者
1、指定topic、partition、offset消費
前面我們在監(jiān)聽消費topic1的時候，監(jiān)聽的是topic1上所有的消息，如果我們想指定topic、指定partition、指定offset來消費呢？也很簡單，@KafkaListener注解已全部為我們提供，
# 設置批量消費 spring.kafka.listener.type=batch # 批量消費每次最多消費多少條消息 spring.kafka.consumer.max-poll-records=50
屬性解釋：
① id：消費者ID；
② groupId：消費組ID；
③ topics：監(jiān)聽的topic，可監(jiān)聽多個；
④ topicPartitions：可配置更加詳細的監(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、批量消費
設置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默認就是方法名，然后我們將這個異常處理器的BeanName放到@KafkaListener注解的errorHandler屬性里面，當監(jiān)聽拋出異常的時候，則會自動調用異常處理器，
@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、消息過濾器
消息過濾器可以在消息抵達consumer之前被攔截，在實際應用中，我們可以根據自己的業(yè)務邏輯，篩選出需要的信息再交由KafkaListener處理，不需要的消息則過濾掉。
配置消息過濾只需要為 監(jiān)聽器工廠 配置一個RecordFilterStrategy（消息過濾策略），返回true的時候消息將會被拋棄，返回false時，消息能正常抵達監(jiān)聽容器。
/**
* @Title 消息轉發(fā)
* @Description 從topic1接收到的消息經過處理后轉發(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";
}
上面實現了一個"過濾奇數、接收偶數"的過濾策略，我們向topic1發(fā)送0-99總共100條消息，看一下監(jiān)聽器的消費情況，可以看到監(jiān)聽器只消費了偶數，
5、消息轉發(fā)
在實際開發(fā)中，我們可能有這樣的需求，應用A從TopicA獲取到消息，經過處理后轉發(fā)到TopicB，再由應用B監(jiān)聽處理消息，即一個應用處理完成后將該消息轉發(fā)至其他應用，完成消息的轉發(fā)。
在SpringBoot集成Kafka實現消息的轉發(fā)也很簡單，只需要通過一個@SendTo注解，被注解方法的return值即轉發(fā)的消息內容，如下，
/**
* @Title 消息轉發(fā)
* @Description 從topic1接收到的消息經過處理后轉發(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)聽器
默認情況下，當消費者項目啟動的時候，監(jiān)聽器就開始工作，監(jiān)聽消費發(fā)送到指定topic的消息，那如果我們不想讓監(jiān)聽器立即工作，想讓它在我們指定的時間點開始工作，或者在我們指定的時間點停止工作，該怎么處理呢——使用KafkaListenerEndpointRegistry，下面我們就來實現：
① 禁止監(jiān)聽器自啟動；
② 創(chuàng)建兩個定時任務，一個用來在指定時間點啟動定時器，另一個在指定時間點停止定時器；
新建一個定時任務類，用注解@EnableScheduling聲明，KafkaListenerEndpointRegistry 在SpringIO中已經被注冊為Bean，直接注入，設置禁止KafkaListener自啟動，
@EnableScheduling
@Component
public class CronTimer {
?
/**
* @KafkaListener注解所標注的方法并不會在IOC容器中被注冊為Bean，
* 而是會被注冊在KafkaListenerEndpointRegistry中，
* 而KafkaListenerEndpointRegistry在SpringIOC中已經被注冊為Bean
**/
@Autowired
private KafkaListenerEndpointRegistry registry;
?
@Autowired
private ConsumerFactory consumerFactory;
?
// 監(jiān)聽器容器工廠(設置禁止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注解后面設置的監(jiān)聽器ID,標識這個監(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("關閉監(jiān)聽器...");
registry.getListenerContainer("timingConsumer").pause();
}
}
啟動項目，觸發(fā)生產者向topic1發(fā)送消息，可以看到consumer沒有消費，因為這時監(jiān)聽器還沒有開始工作，
11:42分監(jiān)聽器啟動開始工作，消費消息，
11：45分監(jiān)聽器停止工作，
到此這篇關于SpringBoot集成kafka全面實戰(zhàn)的文章就介紹到這了,更多相關SpringBoot集成kafka內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
反射機制是在運行狀態(tài)中，對于任意一個類，都能夠知道這個類的所有屬性和方法；對于任意一個對象，都能夠調用它的任意一個方法和屬性；這種動態(tài)獲取的信息以及動態(tài)調用對象的方法的功能稱為java語言的反射機制2017-01-01
SpringBoot多環(huán)境配置及配置文件分類實例詳解
這篇文章主要介紹了SpringBoot多環(huán)境配置及配置文件分類,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-10-10
一個運行中的程序,?產生的對象是大量的,?如果對象不被繼續(xù)使用,?就會成為垃圾,?最后越堆越多,?最后占滿內存,?所以我們要對這些垃圾進行回收,保持程序的正常運行2022-05-05
算法復雜度分為時間復雜度和空間復雜度。其作用： 時間復雜度是度量算法執(zhí)行的時間長短；而空間復雜度是度量算法所需存儲空間的大小2021-11-11
文中詳細講了關于Java哈夫曼樹的概述以及用Java實現的方法,對各位正在學習java數據結構的小伙伴們有很大的幫助喲,需要的朋友可以參考下2021-05-05
這篇文章主要介紹了java鎖synchronized面試常問總結分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-12-12
String.join() 方法是連接指定數組的元素或集合的成員,在每個元素或成員之間使用指定的分隔符,這篇文章主要介紹了String.join()方法示例詳解,需要的朋友可以參考下2024-01-01
區(qū)分java中String+String和String+char
這篇文章主要向大家詳細區(qū)分了java中String+String和String+char，感興趣的小伙伴們可以參考一下2016-01-01
在IDEA中創(chuàng)建跑得起來的Springboot項目
這篇文章主要介紹了在IDEA中創(chuàng)建跑得起來的Springboot項目的圖文教程,需要的朋友可以參考下2018-04-04
這篇文章主要介紹了SpringBoot中使用Servlet的兩種方式小結，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-07-07