更新時間:2023年10月20日10時32分 來源:傳智教育 瀏覽次數(shù):
在Apache Kafka中,消費者(Consumers)和消費者組(Consumer Groups)是核心概念,用于處理消息的訂閱和處理。接下來筆者將詳細解釋它們之間的關系,并提供一個簡單的代碼示例來演示它們的用法。
消費者是Kafka中的客戶端應用程序,它負責訂閱主題并處理從主題中生產(chǎn)的消息。消費者可以獨立訂閱一個或多個主題,并且可以以不同的速度處理消息。它們可以在不同的分區(qū)中并行地處理消息。
消費者組是消費者的邏輯集合,它們一起協(xié)作處理主題中的消息。每個消費者組可以包含一個或多個消費者。消費者組的關鍵特性是它可以協(xié)調(diào)多個消費者來消費主題中的消息,確保每個分區(qū)的消息只被組內(nèi)的一個消費者處理。這有助于實現(xiàn)負載均衡和提高容錯性。
接下來我們看一個使用Java語言的Kafka消費者和消費者組示例:
import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.util.Properties; public class MyKafkaConsumer { public static void main(String[] args) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-broker-list"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName); Consumer<String, String> consumer = new KafkaConsumer<>(props); // 訂閱一個主題 consumer.subscribe(Collections.singletonList("my-topic")); while (true) { // 拉取消息 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 處理消息 System.out.printf("消費者: key=%s, value=%s%n", record.key(), record.value()); } } } }
import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.util.Collections; import java.util.Properties; public class MyKafkaConsumerGroup { public static void main(String[] args) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-broker-list"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // 消費者組名稱 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName); Consumer<String, String> consumer = new KafkaConsumer<>(props); // 訂閱一個主題 consumer.subscribe(Collections.singletonList("my-topic")); while (true) { // 拉取消息 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 處理消息 System.out.printf("消費者組成員: key=%s, value=%s%n", record.key(), record.value()); } } } }
在上述示例中,兩個消費者(可以是同一消費者組的成員)訂閱了同一個主題,但消費者組確保每個分區(qū)的消息只被一個消費者處理,實現(xiàn)了負載均衡和高可用性。這是Kafka中消費者和消費者組的基本關系和用法。