0

我consumer.commit()在 auto.commit=True 模式下运行的消费者的每条消息上调用该方法。当我在本地尝试它时，它没有任何数据丢失或数据重复。

该commit()方法可以给消费者带来什么影响？

class KafkaConsumer(object):
    def __init__(self, topics: list[str], **kwargs: Any):
        config = {
            **kwargs,
        }
        self.consumer = DeserializingConsumer(config)
        self.consumer.subscribe(topics=topics)

    def consume(self, poll_timeout_secs: float = 1.0):
        while True:
            msg = self.consumer.poll(timeout=poll_timeout_secs)
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())
            else:
                yield msg

kafka_consumer = KafkaConsumer(topics=topics, **kwargs) # passed "enable.auto.commit":True

for message in kafka_consumer.consume():
    event = message.value()
    logger.info(f"message {event}")
    kafka_consumer.commit() # what will be the effect of calling this?
4

0 回答 0