以前我們討論的消費(fèi)組，都是 group 的形式，group 可以自動(dòng)地幫助消費(fèi)者分配分區(qū)，且在發(fā)生異常時(shí)，還能自定地進(jìn)行重平衡（Rebalance）。正常來說，group 幫助用戶實(shí)現(xiàn)自動(dòng)監(jiān)聽分區(qū)消費(fèi)，但是在用戶需要指定分區(qū)進(jìn)行精確消費(fèi)的場景下，由于 group 的重平衡機(jī)制，會打破這種消費(fèi)方式，這不前段時(shí)間某項(xiàng)目就有個(gè)需求是這樣的：
消息源端有若干個(gè)，每個(gè)消息源都會產(chǎn)生不同的消息，目標(biāo)端也有若干個(gè)，每個(gè)目標(biāo)端需要消費(fèi)指定的消息源類型。
在以往，由于消費(fèi)組的重平衡機(jī)制會打亂這種消費(fèi)方式，只能申請多個(gè)主題對消息進(jìn)行隔離，每個(gè)消息源將消息發(fā)送到指定主題，目標(biāo)端監(jiān)聽指定的主題。這么做肯定沒有指定分區(qū)消費(fèi)這么優(yōu)雅了，每增加一種消息源，都需要新增一個(gè) topic，且消費(fèi)粒度不能靈活組合。
針對以上問題，Kafka 的提供了獨(dú)立消費(fèi)者模式，可以消費(fèi)者可以指定分區(qū)進(jìn)行消費(fèi)，如果只用一個(gè) topic，每個(gè)消息源啟動(dòng)一個(gè)生產(chǎn)者，分別發(fā)往不同的分區(qū)，消費(fèi)者指定消費(fèi)相關(guān)的分區(qū)即可，用如下圖所示：
但是 Kafka 獨(dú)立消費(fèi)者也有它的限定場景：
1、 Kafka 獨(dú)立消費(fèi)者模式下，Kafka 集群并不會維護(hù)消費(fèi)者的消費(fèi)偏移量，需要每個(gè)消費(fèi)者維護(hù)監(jiān)聽分區(qū)的消費(fèi)偏移量，因此，獨(dú)立消費(fèi)者模式與 group 模式切勿混合使用！
2、group 模式的重平衡機(jī)制在消費(fèi)者異常時(shí)可將其監(jiān)聽的分區(qū)重分配給其它正常的消費(fèi)者，使得這些分區(qū)不會停止被監(jiān)聽消費(fèi)，但是獨(dú)立消費(fèi)者由于是手動(dòng)進(jìn)行監(jiān)聽指定分區(qū)，因此獨(dú)立消費(fèi)者發(fā)生異常時(shí)，并不會將其監(jiān)聽的分區(qū)進(jìn)行重分配，這就會造成某些分區(qū)消息堆積。因此，在該模式下，獨(dú)立消費(fèi)者需要實(shí)現(xiàn)高可用，例如獨(dú)立消費(fèi)者使用 K8s Deployment 進(jìn)行部署。
下面將演示如何使用 Kafka#assgin 方法手動(dòng)訂閱指定分區(qū)進(jìn)行消費(fèi)：
public static void main(String[] args) {
Properties kafkaProperties = new Properties();
kafkaProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.Apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
kafkaProperties.put("bootstrap.servers", "localhost:9092");
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(kafkaProperties);
List<TopicPartition> partitions = new ArrayList<>();
partitions.add(new TopicPartition("test_topic", 0));
partitions.add(new TopicPartition("test_topic", 1));
consumer.assign(partitions); while (true) {
ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(3000));
for (ConsumerRecord<String, byte[]> record : records) {
System.out.printf("topic:%s, partition:%s%n", record.topic(), record.partition());
} }}
