SpringBoot集成kafka全面實戰記錄
本文是SpringBoot+Kafka的實戰講解,如果對kafka的架構原理還不瞭解的讀者,建議先看一下《大白話kafka架構原理》、《秒懂kafka HA(高可用)》兩篇文章。
一、生產者實踐
- 普通生產者
- 帶回調的生產者
- 自定義分區器
- kafka事務提交
二、消費者實踐
- 簡單消費
- 指定topic、partition、offset消費
- 批量消費
- 監聽異常處理器
- 消息過濾器
- 消息轉發
- 定時啟動/停止監聽器
一、前戲
1、在項目中連接kafka,因為是外網,首先要開放kafka配置文件中的如下配置(其中IP為公網IP),
advertised.listeners=PLAINTEXT://112.126.74.249:9092
2、在開始前我們先創建兩個topic:topic1、topic2,其分區和副本數都設置為
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.
當然我們也可以不手動創建topic,在執行代碼kafkaTemplate.send(“topic1”, normalMessage)發送消息時,kafka會幫我們自動完成topic的創建工作,但這種情況下創建的topic默認隻有一個分區,分區也沒有副本。所以,我們可以在項目中新建一個配置類專門用來初始化topic,如下,
@Configuration public class KafkaInitialConfiguration { // 創建一個名為testtopic的Topic並設置分區數為8,分區副本數為2 @Bean public NewTopic initialTopic() { return new NewTopic("testtopic",8, (short) 2 ); } // 如果要修改分區數,隻需修改配置值重啟項目即可 // 修改分區數並不會導致數據的丟失,但是分區數隻能增大不能減小 @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 # 應答級別:多少個分區副本備份完成時向生產者發送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其實就沒用瞭 # 生產端緩沖區大小 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 # 自定義分區器 # 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:重置為分區中最小的offset; # latest:重置為分區中最新的offset(消費分區中新產生的數據); # none:隻要有一個分區不存在已提交的offset,就拋出異常; spring.kafka.consumer.auto-offset-reset=latest # 消費會話超時時間(超過這個時間consumer沒有發送心跳,就會觸發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 # 消費端監聽的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; // 發送消息 @GetMapping("/kafka/normal/{message}") public void sendMessage1(@PathVariable("message") String normalMessage) { kafkaTemplate.send("topic1", normalMessage); } }
2、簡單消費
@Component public class KafkaConsumer { // 消費監聽 @KafkaListener(topics = {"topic1"}) public void onMessage1(ConsumerRecord<?, ?> record){ // 消費的哪個topic、partition的消息,打印出消息內容 System.out.println("簡單消費:"+record.topic()+"-"+record.partition()+"-"+record.value()); } }
上面示例創建瞭一個生產者,發送消息到topic1,消費者監聽topic1消費消息。監聽器用@KafkaListener註解,topics表示監聽的topic,支持同時監聽多個,用英文逗號分隔。啟動項目,postman調接口觸發生產者發送消息,
可以看到監聽器消費成功,
三、生產者
1、帶回調的生產者
kafkaTemplate提供瞭一個回調方法addCallback,我們可以在回調方法中監控消息是否發送成功 或 失敗時做補償處理,有兩種寫法,
@GetMapping("/kafka/callbackOne/{message}") public void sendMessage2(@PathVariable("message") String callbackMessage) { kafkaTemplate.send("topic1", callbackMessage).addCallback(success -> { // 消息發送到的topic String topic = success.getRecordMetadata().topic(); // 消息發送到的分區 int partition = success.getRecordMetadata().partition(); // 消息在分區內的offset long offset = success.getRecordMetadata().offset(); System.out.println("發送消息成功:" + topic + "-" + partition + "-" + offset); }, failure -> { System.out.println("發送消息失敗:" + 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("發送消息失敗:"+ex.getMessage()); } @Override public void onSuccess(SendResult<String, Object> result) { System.out.println("發送消息成功:" + result.getRecordMetadata().topic() + "-" + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset()); } }); }
2、自定義分區器
我們知道,kafka中每個topic被劃分為多個分區,那麼生產者將消息發送到topic時,具體追加到哪個分區呢?這就是所謂的分區策略,Kafka 為我們提供瞭默認的分區策略,同時它也支持自定義分區策略。其路由機制為:
① 若發送消息時指定瞭分區(即自定義分區策略),則直接將消息append到指定分區;
② 若發送消息時未指定 patition,但指定瞭 key(kafka允許為每條消息設置一個key),則對key值進行hash計算,根據計算結果路由到指定分區,這種情況下可以保證同一個 Key 的所有消息都進入到相同的分區;
③ patition 和 key 都未指定,則使用kafka默認的分區策略,輪詢選出一個 patition;
※ 我們來自定義一個分區策略,將消息發送到我們指定的partition,首先新建一個分區器類實現Partitioner接口,重寫方法,其中partition方法的返回值就表示將消息發送到幾號分區,
public class CustomizePartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 自定義分區規則(這裡假設全部發到0號分區) // ...... return 0; } @Override public void close() { } @Override public void configure(Map<String, ?> configs) { } }
在application.propertise中配置自定義分區器,配置的值就是分區器類的全路徑名,
# 自定義分區器 spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
3、kafka事務提交
如果在發送消息時需要創建事務,可以使用 KafkaTemplate 的 executeInTransaction 方法來聲明事務,
@GetMapping("/kafka/transaction") public void sendMessage7(){ // 聲明事務:後面報錯消息不會發出去 kafkaTemplate.executeInTransaction(operations -> { operations.send("topic1","test executeInTransaction"); throw new RuntimeException("fail"); }); // 不聲明事務:後面報錯但前面消息已經發送成功瞭 kafkaTemplate.send("topic1","test executeInTransaction"); throw new RuntimeException("fail"); }
四、消費者
1、指定topic、partition、offset消費
前面我們在監聽消費topic1的時候,監聽的是topic1上所有的消息,如果我們想指定topic、指定partition、指定offset來消費呢?也很簡單,@KafkaListener註解已全部為我們提供,
# 設置批量消費 spring.kafka.listener.type=batch # 批量消費每次最多消費多少條消息 spring.kafka.consumer.max-poll-records=50
屬性解釋:
① id:消費者ID;
② groupId:消費組ID;
③ topics:監聽的topic,可監聽多個;
④ topicPartitions:可配置更加詳細的監聽信息,可指定topic、parition、offset監聽。
上面onMessage2監聽的含義:監聽topic1的0號分區,同時監聽topic2的0號分區和topic2的1號分區裡面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來接收,監聽代碼如下,
// 新建一個異常處理器,用@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在消費時發生的異常。
新建一個 ConsumerAwareListenerErrorHandler 類型的異常處理方法,用@Bean註入,BeanName默認就是方法名,然後我們將這個異常處理器的BeanName放到@KafkaListener註解的errorHandler屬性裡面,當監聽拋出異常的時候,則會自動調用異常處理器,
@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; } // 消息過濾監聽 @KafkaListener(topics = {"topic1"},containerFactory = "filterContainerFactory") public void onMessage6(ConsumerRecord<?, ?> record) { System.out.println(record.value()); } }
執行看一下效果,
4、消息過濾器
消息過濾器可以在消息抵達consumer之前被攔截,在實際應用中,我們可以根據自己的業務邏輯,篩選出需要的信息再交由KafkaListener處理,不需要的消息則過濾掉。
配置消息過濾隻需要為 監聽器工廠 配置一個RecordFilterStrategy(消息過濾策略),返回true的時候消息將會被拋棄,返回false時,消息能正常抵達監聽容器。
/** * @Title 消息轉發 * @Description 從topic1接收到的消息經過處理後轉發到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發送0-99總共100條消息,看一下監聽器的消費情況,可以看到監聽器隻消費瞭偶數,
5、消息轉發
在實際開發中,我們可能有這樣的需求,應用A從TopicA獲取到消息,經過處理後轉發到TopicB,再由應用B監聽處理消息,即一個應用處理完成後將該消息轉發至其他應用,完成消息的轉發。
在SpringBoot集成Kafka實現消息的轉發也很簡單,隻需要通過一個@SendTo註解,被註解方法的return值即轉發的消息內容,如下,
/** * @Title 消息轉發 * @Description 從topic1接收到的消息經過處理後轉發到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、定時啟動、停止監聽器
默認情況下,當消費者項目啟動的時候,監聽器就開始工作,監聽消費發送到指定topic的消息,那如果我們不想讓監聽器立即工作,想讓它在我們指定的時間點開始工作,或者在我們指定的時間點停止工作,該怎麼處理呢——使用KafkaListenerEndpointRegistry,下面我們就來實現:
① 禁止監聽器自啟動;
② 創建兩個定時任務,一個用來在指定時間點啟動定時器,另一個在指定時間點停止定時器;
新建一個定時任務類,用註解@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; // 監聽器容器工廠(設置禁止KafkaListener自啟動) @Bean public ConcurrentKafkaListenerContainerFactory delayContainerFactory() { ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory(); container.setConsumerFactory(consumerFactory); //禁止KafkaListener自啟動 container.setAutoStartup(false); return container; } // 監聽器 @KafkaListener(id="timingConsumer",topics = "topic1",containerFactory = "delayContainerFactory") public void onMessage1(ConsumerRecord<?, ?> record){ System.out.println("消費成功:"+record.topic()+"-"+record.partition()+"-"+record.value()); } // 定時啟動監聽器 @Scheduled(cron = "0 42 11 * * ? ") public void startListener() { System.out.println("啟動監聽器..."); // "timingConsumer"是@KafkaListener註解後面設置的監聽器ID,標識這個監聽器 if (!registry.getListenerContainer("timingConsumer").isRunning()) { registry.getListenerContainer("timingConsumer").start(); } //registry.getListenerContainer("timingConsumer").resume(); } // 定時停止監聽器 @Scheduled(cron = "0 45 11 * * ? ") public void shutDownListener() { System.out.println("關閉監聽器..."); registry.getListenerContainer("timingConsumer").pause(); } }
啟動項目,觸發生產者向topic1發送消息,可以看到consumer沒有消費,因為這時監聽器還沒有開始工作,
11:42分監聽器啟動開始工作,消費消息,
11:45分監聽器停止工作,
到此這篇關於SpringBoot集成kafka全面實戰的文章就介紹到這瞭,更多相關SpringBoot集成kafka內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Java分佈式學習之Kafka消息隊列
- SpringBoot集成Kafka的步驟
- Java實現Kafka生產者和消費者的示例
- 解決kafka消息堆積及分區不均勻的問題
- 深入研究spring boot集成kafka之spring-kafka底層原理