Kafka消費(fèi)者訂閱方式

Kafka為消費(fèi)者提供了三種類型的訂閱消費(fèi)方式：訂閱主題集合、正則表達(dá)式訂閱主題、訂閱指定主題的分區(qū)集合。三種方式只能使用其中一種。

1.指定主題消費(fèi)

一個消費(fèi)者可以使用KafkaConsumer提供的subscribe()方法訂閱一個或多個主題，訂閱主題集合和正則表達(dá)式訂閱主題都使用此方法實現(xiàn)的。下面兩種方式都可以訂閱topic_1120主題。

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 訂閱主題
consumer.subscribe(Collections.singletonList("topic_1120"));

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//正則表達(dá)式.*代表后續(xù)0個或者多個任意字符。
consumer.subscribe(Pattern.compile("topic.*"));

訂閱主題在源碼中由4個方法重載實現(xiàn)，其中兩個帶listener的方法是可以自定義Rebalance重平衡的監(jiān)聽類。

@Override
public void subscribe(Collection<String> topics) {
? ? subscribe(topics, new NoOpConsumerRebalanceListener());
}
@Override
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
? ? //省略源碼
}
@Override
public void subscribe(Pattern pattern) {
? ? subscribe(pattern, new NoOpConsumerRebalanceListener());
}
@Override
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
? ? //省略源碼
}

2.指定分區(qū)消費(fèi)

消費(fèi)者指定分區(qū)消費(fèi)是通過KafkaConsumer提供的assign()方法實現(xiàn)的，assign()方法入?yún)镃ollection, 其中TopicPartition有2個屬性, topic和partition, 分區(qū)從0開始編號。使用assign()方法訂閱指定主題test_1120分區(qū)0的消息。

/訂閱指定分區(qū)
consumer.assign(Collections.singleton(new TopicPartition("topic_1120", 0)));

3.取消訂閱

取消訂閱調(diào)用unsubscribe()方法。

consumer.unsubscribe();

小結(jié)：subscribe()具有自動重平衡的功能，來實現(xiàn)消費(fèi)負(fù)載均衡和故障自動轉(zhuǎn)移，而assign()不具備這種功能。

Kafka概述

定義

Kafka是一個分布式的基于發(fā)布/訂閱模式的消息隊列，主要應(yīng)用于大數(shù)據(jù)實時處理領(lǐng)域。

消息隊列

1.傳統(tǒng)消息隊列的應(yīng)用場景

使用消息隊列的好處

1）解耦

允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。

2）可恢復(fù)性

系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。消息隊列降低了進(jìn)程間的耦合度，所 以即使一個處理消息的進(jìn)程掛掉，加入隊列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。

3）緩沖

有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。

4）靈活性 & 峰值處理能力

在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。 如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時待命無疑是巨大的浪費(fèi)。使用消息隊列 能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會因為突發(fā)的超負(fù)荷的請求而完全崩潰。

5）異步通信

很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機(jī)制，允許用戶 把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要 的時候再去處理它們。

2.消息隊列的兩種模式

（1）點(diǎn)對點(diǎn)模式（一對一，消費(fèi)者主動拉取數(shù)據(jù)，消息收到后消息清除）

消息生產(chǎn)者生產(chǎn)消息發(fā)送到Queue中，然后消息消費(fèi)者從Queue中取出并且消費(fèi)消息。 消息被消費(fèi)以后，queue 中不再有存儲，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。 Queue 支持存在多個消費(fèi)者，但是對一個消息而言，只會有一個消費(fèi)者可以消費(fèi)。

（2）發(fā)布/訂閱模式（一對多，消費(fèi)者消費(fèi)數(shù)據(jù)之后不會清除消息）

消息生產(chǎn)者（發(fā)布）將消息發(fā)布到 topic 中，同時有多個消息消費(fèi)者（訂閱）消費(fèi)該消 息。和點(diǎn)對點(diǎn)方式不同，發(fā)布到 topic 的消息會被所有訂閱者消費(fèi)。

Kafka 基礎(chǔ)架構(gòu)

• Producer ：消息生產(chǎn)者，就是向 kafka broker 發(fā)消息的客戶端；
• Consumer ：消息消費(fèi)者，向 kafka broker 取消息的客戶端；
• Consumer Group （CG）：消費(fèi)者組，由多個 consumer 組成。消費(fèi)者組內(nèi)每個消費(fèi)者負(fù) 責(zé)消費(fèi)不同分區(qū)的數(shù)據(jù)，一個分區(qū)只能由一個組內(nèi)消費(fèi)者消費(fèi)；消費(fèi)者組之間互不影響。所 有的消費(fèi)者都屬于某個消費(fèi)者組，即消費(fèi)者組是邏輯上的一個訂閱者。
• Broker ：一臺 kafka 服務(wù)器就是一個 broker。一個集群由多個 broker 組成。一個 broker 可以容納多個 topic。
• Topic ：可以理解為一個隊列，生產(chǎn)者和消費(fèi)者面向的都是一個 topic；
• Partition：為了實現(xiàn)擴(kuò)展性，一個非常大的 topic 可以分布到多個 broker（即服務(wù)器）上， 一個 topic 可以分為多個 partition，每個 partition 是一個有序的隊列；
• Replica：副本，為保證集群中的某個節(jié)點(diǎn)發(fā)生故障時，該節(jié)點(diǎn)上的 partition 數(shù)據(jù)不丟失， 且 kafka 仍然能夠繼續(xù)工作，kafka 提供了副本機(jī)制，一個 topic 的每個分區(qū)都有若干個副本， 一個 leader 和若干個 follower。
• leader：每個分區(qū)多個副本的“主”，生產(chǎn)者發(fā)送數(shù)據(jù)的對象，以及消費(fèi)者消費(fèi)數(shù)據(jù)的對 象都是 leader。
• follower：每個分區(qū)多個副本中的“從”，實時從 leader 中同步數(shù)據(jù)，保持和 leader 數(shù)據(jù) 的同步。leader 發(fā)生故障時，某個 follower 會成為新的 follower。

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論