kafka-python：消费者对象可以在不订阅任何主题的情况下加入消费者组吗？答案

【问题标题】：kafka-python: Can a consumer object join a consumer group without subscribing to any topics?kafka-python：消费者对象可以在不订阅任何主题的情况下加入消费者组吗？
【发布时间】：2021-05-04 09:45:19
【问题描述】：

我有一个topics_subscriber.py，它会不断检查任何新主题并订阅它们。

def _subscribe_topics(consumer: KafkaConsumer) -> None:
    topics_to_subscribe = {
        topic
        for topic in consumer.topics()
        if topic.startswith("topic-")
    }

    subscribed_topics = consumer.subscription()
    print(subscribed_topics)

    new_topics = (
        topics_to_subscribe - subscribed_topics
        if subscribed_topics
        else topics_to_subscribe
    )

    if new_topics:
        print("new topics detected:")
        for topic in new_topics:
            print(topic)

        print("\nsubscribing to old+new topics\n")
        consumer.subscribe(topics=list(topics_to_subscribe)) // subscribe() is not incremental, hence subscribing to old+new topics 

        print(consumer.subscription())
    else:
        print("\nno new topics detected: exiting normally\n")


def main() -> None:
    consumer = kafka.KafkaConsumer(
                   client_id="client_1",
                   group_id="my_group",
                   bootstrap_servers=['localhost:9092']
               )

    while True:
        _subscribe_topics(consumer)

        print("\nsleeping 10 sec\n")
        time.sleep(10)

现在，在另一个脚本kafka_extractor.py 中，我想创建一个新的消费者并加入my_group 消费者组，并开始使用该组订阅的主题中的消息。即没有专门为这个新消费者订阅主题。

def main() -> None:
    consumer = kafka.KafkaConsumer(
                   client_id="client_2",
                   group_id="my_group",
                   bootstrap_servers=['localhost:9092']
               )
    print("created consumer")
    print(consumer.subscription())

    for msg in consumer:
        print(msg.topic)

kafka_extractor.py 的输出中需要注意两点：

print(consumer.subscription()) 输出为无
for msg in consumer: -> 卡住了，不前进也不退出程序。

任何方向都会在这里受到赞赏。

【问题讨论】：

标签： python apache-kafka kafka-consumer-api subscribe kafka-python

【解决方案1】：

将消费者加入群组不会获得群组现有的主题，因为群组可以包含多个主题，消费者实例通过订阅来决定消费哪些主题

您的循环卡住了，因为它默认从订阅主题的末尾读取，但没有

如果您想使用定期刷新的主题列表，请使用正则表达式订阅

【讨论】：