FastKafka库：enable_autocommit=False时如何手动提交消费消息？

要实现仅在消息处理成功时手动提交偏移量，需要从关闭自动提交、注入消费者实例、业务逻辑异常捕获三个环节入手，具体步骤如下：

1. 初始化FastKafka应用时关闭自动提交 #

在创建FastKafka实例时，通过consumer_config配置关闭自动提交，并设置偏移量重置策略：

from fastkafka import FastKafka
from pydantic import BaseModel

# 定义消息模型
class BusinessMessage(BaseModel):
    data: str

# 初始化Kafka应用，配置消费者参数
kafka_app = FastKafka(
    kafka_brokers={"default": "localhost:9092"},
    consumer_config={
        "enable_auto_commit": False,  # 全局关闭自动提交
        "auto_offset_reset": "earliest",  # 消费者重启时从最早未提交的偏移量开始拉取
        "group_id": "my_consumer_group",  # 必须指定消费组ID
    }
)

2. 编写带手动提交逻辑的消息处理器 #

使用consumes装饰器时，显式设置auto_commit=False，并在处理函数中注入consumer实例，通过异常捕获控制提交时机：

# 你的业务处理函数，处理失败会抛出异常
def process_business_data(msg_data: str) -> None:
    if len(msg_data) < 5:
        raise ValueError("消息数据长度不符合要求")
    # 这里写具体的业务处理逻辑
    print(f"成功处理业务数据: {msg_data}")

@kafka_app.consumes(topic="business_topic", auto_commit=False)
async def handle_business_message(msg: BusinessMessage, consumer) -> None:
    try:
        # 调用业务处理函数
        process_business_data(msg.data)
        # 处理无异常，手动提交当前偏移量
        await consumer.commit()
        print("消息偏移量已提交")
    except Exception as e:
        # 处理失败，不提交偏移量，消费者会在下一次拉取时重新获取该消息
        print(f"消息处理失败，不提交偏移量: {str(e)}")

关键说明 #

• 双重关闭自动提交：全局consumer_config和consumes装饰器的auto_commit=False要同时设置，避免自动提交逻辑生效。
• 消费者实例注入：处理函数中声明consumer参数，FastKafka会自动注入当前的消费者实例，用于调用commit()方法。
• 精确提交控制：如果需要针对单条消息精确提交偏移量，可以替换commit()为：

  await consumer.commit(
      {msg.topic: {msg.partition: msg.offset + 1}}
  )
  

• 异常处理策略：处理失败时可以根据业务需求选择不提交（让消息重发）、记录错误日志或死信队列存储等操作。

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