0

当我设置msg = consumer.poll(timeout=10.0)消费者等待 10 秒并按None预期返回时，但是当我将其更改msg = consumer.poll(timeout=3600.0)为此消费者时，只需立即返回None，而不是按预期等待 3600 秒。我在这里错过了什么吗？如果需要，这里是完整的代码。

running = True
conf = {'bootstrap.servers': bootstrap_servers,
        'group.id': 'foo',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
        'on_commit': commit_completed}
consumer = Consumer(conf)


def msg_process(msg):
    print(f"key: {msg.key().decode('utf-8')}, value: {msg.value().decode('utf-8')}")


def basic_consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        msg_count = 0
        while running:
            msg = consumer.poll(timeout=3600.0)
            if msg is None:
                print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: no new message")
                continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    print('%% %s [%d] reached end at offset %d\n' %
                          (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                # consumer.commit(async=False)
                msg_process(msg)
                msg_count += 1
                if msg_count % MIN_COMMIT_COUNT == 0:
                    consumer.commit(async=True)
    finally:
        # Close down consumer to commit final offsets.
        consumer.close()


def shutdown():
    running = False


basic_consume_loop(consumer, [topic_user])
4

1 回答 1

0

可能是因为 fetch.max.wait.ms设置低于传入的值poll()。

于 2021-12-23T22:15:19.507 回答