如何在FastAPI中处理AIOKafka getmany()的Kafka停服异常？

当前代码中，consumer.getmany(timeout_ms=10000)在Kafka停服时只会超时返回空的data，不会主动抛出异常，导致循环持续执行、不断打印"Try consume"，无法进入except块触发错误日志。这是因为AIOKafka消费者默认会尝试重新连接Kafka，在重试期间getmany仅会超时返回空结果，不会直接抛出连接异常。

1. 调整消费者配置，加速连接异常检测 #

在create_consumer函数中添加以下配置，让消费者更快感知Kafka连接断开：

def create_consumer():
    return AIOKafkaConsumer(
        # 你的原有配置（如topic、group_id等）
        bootstrap_servers=settings.kafka.bootstrap_servers,
        group_id=settings.kafka.group_id,
        enable_auto_commit=False,  # 保持手动提交配置
        metadata_max_age_ms=30000,  # 缩短元数据刷新间隔，更快发现Broker不可用
        retry_backoff_ms=1000,  # 重试间隔
        max_retries=3,  # 超过最大重试次数后抛出异常
        api_version_auto_timeout_ms=5000,  # 缩短API版本协商超时时间
    )

2. 修改consume函数，捕获特定异常并检测连接状态 #

更新消费逻辑，明确捕获AIOKafka连接类异常，并通过空数据计数器主动识别连接异常：

from aiokafka.errors import KafkaError, KafkaConnectionError, KafkaTimeoutError

async def consume(db: Session = next(get_db())):
    """Consume and process messages from Kafka."""
    empty_count = 0
    max_empty_threshold = 3  # 连续3次获取空数据则判定连接异常
    
    while True:
        try:
            print("Try consume")
            data = await consumer.getmany(timeout_ms=10000)
            
            if not data:
                empty_count += 1
                if empty_count >= max_empty_threshold:
                    # 主动检查消费者连接状态
                    if not consumer._client.is_connected():
                        raise KafkaConnectionError("Lost connection to Kafka broker")
            else:
                empty_count = 0  # 获取到数据后重置计数器
                for tp, msgs in data.items():
                    if msgs:
                        for msg in msgs:
                            await process_message(msg, db)
                        await consumer.commit({tp: msgs[-1].offset+1})
                        
        except (KafkaConnectionError, KafkaTimeoutError, KafkaError) as e:
            # 打印ERROR LOG
            global_items["logger"].error(f"Kafka consume error: {str(e)}")
            # 异常后等待5秒再尝试
            await asyncio.sleep(5)
        except Exception as e:
            global_items["logger"].error(f"Unexpected consume error: {str(e)}")
        finally:
            await asyncio.sleep(settings.common.consumer_pause_sec)

3. 优化启动任务管理（可选） #

确保asyncio.gather的任务状态可监控：

@app.on_event("startup")
async def startup_event():
    """Start up event for FastAPI application."""
    global_items["logger"].info("Starting up...")
    await consumer.start()

    # 保存任务实例，方便后续监控
    startup_tasks = asyncio.gather(
        consume(),
        # 其他任务
        return_exceptions=True
    )
    # 添加任务完成回调
    startup_tasks.add_done_callback(lambda task: global_items["logger"].info(f"Startup tasks completed: {task.result()}"))

• 缩短元数据刷新间隔和设置重试阈值，让消费者在Kafka停服后更快触发连接异常。
• 通过空数据计数器，在连续多次获取不到数据时主动检查连接状态，触发异常捕获逻辑。
• 明确捕获AIOKafka特定异常类型，避免通用Exception捕获无关错误。

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