如何在Python中定时X分钟后关闭Kafka Consumer？

首先明确consumer_timeout_ms的生效逻辑：仅当消费者在指定时长内未拉取到任何消息时，才会触发超时退出。如果期间持续有消息流入，该参数不会生效。以下是几种可靠的超时关闭方案：

方案1：使用定时器强制关闭 #

通过threading.Timer在指定时间后强制调用消费者的close()方法，不受消息是否存在影响：

import threading
from kafka import KafkaConsumer

def close_consumer(consumer):
    print("超时，关闭消费者")
    consumer.close()

# 初始化消费者
consumer = KafkaConsumer(bootstrap_servers='xxx:xx',
                     group_id=mygroupid,
                     security_protocol='SSL',
                     ssl_check_hostname=True,
                     ssl_cafile='cccc.pem',
                     ssl_certfile='sss.pem',
                     ssl_keyfile='kk.pem',
                     auto_offset_reset='latest')

# 设置60秒后关闭消费者
timer = threading.Timer(60.0, close_consumer, args=[consumer])
timer.start()

# 消费消息循环
try:
    for msg in consumer:
        # 处理消息，写入文件逻辑
        with open('output.txt', 'a') as f:
            f.write(f"{msg.value.decode()}\n")
        # 如果消费者已关闭，退出循环
        if not consumer.bootstrap_connected():
            break
finally:
    timer.cancel()  # 提前退出时取消定时器
    consumer.close()

方案2：记录启动时间，循环内检查超时 #

在消费循环中实时检查当前时间与启动时间的差值，达到指定时长就主动退出循环并关闭消费者：

import time
from kafka import KafkaConsumer

# 初始化消费者
consumer = KafkaConsumer(bootstrap_servers='xxx:xx',
                     group_id=mygroupid,
                     security_protocol='SSL',
                     ssl_check_hostname=True,
                     ssl_cafile='cccc.pem',
                     ssl_certfile='sss.pem',
                     ssl_keyfile='kk.pem',
                     auto_offset_reset='latest')

timeout_seconds = 60
start_time = time.time()

# 消费消息循环
try:
    while True:
        # 检查是否超时
        if time.time() - start_time > timeout_seconds:
            print("超时，退出消费")
            break
        # 拉取消息，设置短轮询时间
        messages = consumer.poll(timeout_ms=1000)
        for topic_partition, records in messages.items():
            for msg in records:
                # 处理消息，写入文件
                with open('output.txt', 'a') as f:
                    f.write(f"{msg.value.decode()}\n")
finally:
    consumer.close()

排查原参数未生效的可能原因 #

• 消费代码是否进入了消费循环？consumer_timeout_ms仅在调用poll()或迭代消费者时才会触发检查。
• 超时期间是否一直有消息产生？如果持续有消息，该参数不会触发退出。

内容的提问来源于stack exchange，提问作者dev_code