FastAPI中如何让Kafka消费者在独立线程运行以启用HTTP接口

问题根源 #

你在FastAPI的startup事件中直接调用await consume()，而consume是一个包含无限循环的异步函数，会持续占用asyncio事件循环，导致FastAPI无法处理后续的HTTP请求，所以你的接口无法被正常调用。

解决方案 #

使用asyncio.create_task将Kafka消费任务作为后台异步任务启动，这样事件循环可以同时处理HTTP请求和Kafka消费逻辑，不会被阻塞。

修改后的完整代码 #

# main.py
from pathlib import Path
from fastapi import FastAPI, APIRouter, HTTPStatus  # 补充导入HTTPStatus
from app.api.v1.api import api_router
from app.core.config import settings
import logging
import asyncio  # 新增asyncio导入
from aiokafka import AIOKafkaConsumer
import hashlib

log = logging.getLogger("uvicorn")

router = APIRouter(prefix="/kafka_consumer")

# 用于保存后台消费任务，方便shutdown时取消
consumer_task = None

async def consume():
    """Consume and print messages from Kafka."""
    try:
        async for msg in consumer:
            print("TODO: process msg ", msg.value.decode())
    except asyncio.CancelledError:
        # 捕获任务取消信号，优雅退出
        log.info("Kafka consumer task cancelled")

async def check_kafka() -> bool:
    """
    Checks if Kafka is available by fetching all metadata from the Kafka client.

    Returns:
        bool: True if Kafka is available, False otherwise.
    """
    try:
        await consumer.client.fetch_all_metadata()
    except Exception as exc:
        logging.error(f'Kafka is not available: {exc}')
        return False  # 补充返回False分支
    else:
        return True

@router.get("/healthz/ready")
async def ready_check():
    """
    Check the health of application dependencies.

    Returns:
        HTTP status code with corresponding message.
    """
    # 注意：http_client需提前定义或导入，若未实现可暂时注释
    readiness_probes = await asyncio.gather(
        check_kafka(),
        # http_client.is_ready(),
        return_exceptions=True  # 防止单个探针失败导致整个检查崩溃
    )
    ready = all(probe for probe in readiness_probes)
    if ready:
        return {"status": "ready"}, HTTPStatus.OK
    else:
        return {"status": "not ready"}, HTTPStatus.SERVICE_UNAVAILABLE

def create_application() -> FastAPI:
    """Create FastAPI application and set routes.

    Returns:
        FastAPI: The created FastAPI instance.
    """
    application = FastAPI(openapi_url="/kafka_consumer/openapi.json", docs_url="/kafka_consumer/docs")
    application.include_router(router, tags=["consumer"])
    return application

def create_consumer() -> AIOKafkaConsumer:
    """Create AIOKafkaConsumer.

    Returns:
        AIOKafkaConsumer: The created AIOKafkaConsumer instance.
    """
    return AIOKafkaConsumer(
        settings.kafka_topics,
        bootstrap_servers=settings.kafka_instance,
    )

app = create_application()
consumer = create_consumer()

@app.on_event("startup")
async def startup_event():
    """Start up event for FastAPI application."""
    global consumer_task
    log.info("Starting up...")
    await consumer.start()
    log.info("Starting Kafka consumer task in background")
    # 将消费任务作为后台任务启动，不阻塞事件循环
    consumer_task = asyncio.create_task(consume())
    log.info("Application is ready to handle HTTP requests")

@app.on_event("shutdown")
async def shutdown_event():
    """Shutdown event for FastAPI application."""
    global consumer_task
    log.info("Shutting down...")
    # 取消后台消费任务并等待其结束
    if consumer_task:
        consumer_task.cancel()
        await consumer_task
    await consumer.stop()

关键修改点说明 #

1. 新增asyncio导入：用于创建和管理后台异步任务
2. 后台任务启动：用asyncio.create_task(consume())替代await consume()，让消费任务在事件循环中并发执行，不阻塞主线程的HTTP请求处理
3. 优雅关闭：保存consumer_task对象，在shutdown事件中取消任务并等待其结束，确保Kafka连接等资源被正确释放
4. 细节修复：补充check_kafka函数的return False分支，调整ready_check的返回格式使其更符合FastAPI规范

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