是的,Kafka的消費者組ID(group ID)可以手動指定。在創建消費者時,可以通過設置group.id
屬性來指定消費者所屬的消費者組ID。這個屬性是可選的,如果未指定,Kafka會為該消費者生成一個默認的消費者組ID。
以下是一個使用Python的confluent_kafka
庫創建Kafka消費者的示例,其中手動指定了消費者組ID:
from confluent_kafka import Consumer, KafkaError
conf = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'my_custom_group'
}
consumer = Consumer(conf)
consumer.subscribe(['my_topic'])
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
print(f"End of partition reached {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
else:
raise KafkaException(msg.error())
else:
print(f"Received message: {msg.value().decode('utf-8')}")
except KeyboardInterrupt:
pass
finally:
consumer.close()
在這個示例中,我們將group.id
設置為my_custom_group
,以便在Kafka集群中識別該消費者所屬的消費者組。