首頁猿問何時(shí)使用...

何時(shí)使用 ConcurrentKafkaListenerContainerFactory？

Java

臨摹微笑 2022-07-20 19:26:12

我是 kafka 的新手，我瀏覽了文檔，但我什么都不懂。有人可以解釋何時(shí)使用該ConcurrentKafkaListenerContainerFactory課程嗎？我已經(jīng)使用了該Kafkaconsumer課程，但我看到ConcurrentKafkaListenerContainerFactory在我當(dāng)前的項(xiàng)目中使用了該課程。請(qǐng)解釋它的用途。

查看完整描述

2 回答

翻閱古今

TA貢獻(xiàn)1780條經(jīng)驗(yàn) 獲得超5個(gè)贊

Kafka 消費(fèi)者不是線程安全的。所有網(wǎng)絡(luò) I/O 都發(fā)生在進(jìn)行調(diào)用的應(yīng)用程序的線程中。確保多線程訪問正確同步是用戶的責(zé)任。非同步訪問將導(dǎo)致ConcurrentModificationException.

如果消費(fèi)者被分配了多個(gè)分區(qū)來獲取數(shù)據(jù)，它將嘗試同時(shí)從所有分區(qū)中消費(fèi)，從而有效地為這些分區(qū)提供相同的消費(fèi)優(yōu)先級(jí)。但是，在某些情況下，消費(fèi)者可能希望首先專注于從分配的分區(qū)的某個(gè)子集全速獲取，并且僅在這些分區(qū)幾乎沒有或沒有數(shù)據(jù)要消耗時(shí)才開始獲取其他分區(qū)。

春卡夫卡

ConcurrentKafkaListenerContainerFactory用于為帶注釋的方法創(chuàng)建容器@KafkaListener

MessageListenerContainer春天卡夫卡有兩個(gè)

KafkaMessageListenerContainer

ConcurrentMessageListenerContainer

KafkaMessageListenerContainer接收來自單個(gè)線程上所有主題或分區(qū)的所有消息。ConcurrentMessageListenerContainer委托給一個(gè)或多個(gè)實(shí)例KafkaMessageListenerContainer以提供多線程消費(fèi)。

使用 ConcurrentMessageListenerContainer

@Bean

KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>

kafkaListenerContainerFactory() {

ConcurrentKafkaListenerContainerFactory<Integer, String> factory =

new ConcurrentKafkaListenerContainerFactory<>();

factory.setConsumerFactory(consumerFactory());

factory.setConcurrency(3);

factory.getContainerProperties().setPollTimeout(3000);

return factory;

}

它具有并發(fā)屬性。例如， container.setConcurrency(3) 創(chuàng)建了三個(gè)KafkaMessageListenerContainer實(shí)例。

如果TopicPartition提供了6個(gè)實(shí)例，并發(fā)為3；每個(gè)容器有兩個(gè)分區(qū)。對(duì)于五個(gè) TopicPartition 實(shí)例，兩個(gè)容器獲得兩個(gè)分區(qū)，第三個(gè)獲得一個(gè)。如果并發(fā)大于 TopicPartition 的數(shù)量，則將并發(fā)調(diào)低，使每個(gè)容器獲得一個(gè)分區(qū)。

反對(duì) 回復(fù) 2022-07-20