python写一段Kafka消费者手动提交的代码

以下是一个简单的 Kafka 消费者手动提交的 Python 代码示例：

from kafka import KafkaConsumer
from kafka.errors import KafkaError

consumer = KafkaConsumer('my_topic',
                         bootstrap_servers=['localhost:9092'],
                         group_id='my_group',
                         auto_offset_reset='earliest',
                         enable_auto_commit=False) # 禁止自动提交

try:
    for message in consumer:
        # 消费消息
        print(message.value)

        # 手动提交偏移量
        consumer.commit()
except KafkaError as e:
    print("Error: ", e)
finally:
    consumer.close()

在上面的代码示例中，我们创建了一个 Kafka 消费者对象 `consumer`，并设置了以下参数：

- `bootstrap_servers`：指定 Kafka 集群中的 broker 地址；
- `group_id`：指定消费者组的名称；
- `auto_offset_reset`：当消费者第一次连接到 Kafka 集群时，需要确定从哪个偏移量开始消费。这里设置为 `earliest` 表示从最早的偏移量开始消费；
- `enable_auto_commit`：设置为 `False`，禁止自动提交偏移量。

在消费消息时，我们可以通过 `consumer.commit()` 方法手动提交偏移量。这样可以确保消费者在处理完消息后再提交偏移量，避免出现消息重复消费或漏消费的情况。在 `try-except-finally` 语句块中，我们使用 `consumer.close()` 方法关闭了消费者对象。