优化FastAPI高内存缓存的多进程扩展：事件驱动架构实践

pip install celery redis

from celery import Celery

# 配置Celery，使用Redis作为消息代理和结果存储
celery_app = Celery(
    'my_fastapi_tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

# 定义一个模拟的耗时任务，它可能需要访问“缓存”数据
@celery_app.task
def process_huge_data_task(data_id: str):
    """
    模拟处理大量数据的任务。
    这个任务将由Celery Worker在独立的进程中执行。
    如果需要访问共享数据，可以考虑将数据ID传递给Worker，
    Worker再从一个共享的、独立于Web服务器的存储（如分布式缓存或数据库）中获取。
    """
    print(f"Celery Worker 正在处理数据: {data_id}")
    # 假设这里是访问和处理8GB数据的逻辑
    import time
    time.sleep(10) # 模拟耗时操作
    result = f"数据 {data_id} 处理完成。"
    print(result)
    return result

from fastapi import FastAPI, BackgroundTasks
from app.celery_app import process_huge_data_task

app = FastAPI()

@app.get("/process_data/{data_id}")
async def trigger_data_processing(data_id: str):
    # 将耗时任务发送给Celery Worker异步处理
    task = process_huge_data_task.delay(data_id)
    # 立即返回响应，包含任务ID
    return {"message": "数据处理任务已提交", "task_id": task.id}

@app.get("/task_status/{task_id}")
async def get_task_status(task_id: str):
    task = process_huge_data_task.AsyncResult(task_id)
    if task.ready():
        return {"status": "完成", "result": task.result}
    elif task.pending:
        return {"status": "等待中"}
    elif task.failed():
        return {"status": "失败", "error": str(task.result)}
    else:
        return {"status": "进行中"}

from fastapi import FastAPI
# 假设你有一个消息队列客户端，例如 for Kafka: confluent-kafka-python
# from confluent_kafka import Producer

app = FastAPI()
# producer = Producer({'bootstrap.servers': 'localhost:9092'}) # Kafka Producer

@app.post("/submit_analysis")
async def submit_analysis(payload: dict):
    # 将分析请求发布到消息队列
    # producer.produce('data_analysis_topic', value=json.dumps(payload).encode('utf-8'))
    # producer.flush()
    print(f"分析请求已发布到消息队列: {payload}")
    return {"message": "分析请求已提交到队列"}

# 这是一个独立的Python服务，运行在另一个进程或服务器上
# from confluent_kafka import Consumer, KafkaException

# consumer = Consumer({
#     'bootstrap.servers': 'localhost:9092',
#     'group.id': 'my_analysis_group',
#     'auto.offset.reset': 'earliest'
# })
# consumer.subscribe(['data_analysis_topic'])

# while True:
#     msg = consumer.poll(timeout=1.0)
#     if msg is None: continue
#     if msg.error():
#         if msg.error().code() == KafkaException._PARTITION_EOF:
#             continue
#         else:
#             print(msg.error())
#             break
#     
#     data_to_process = json.loads(msg.value().decode('utf-8'))
#     print(f"消费者正在处理数据: {data_to_process}")
#     # 在这里执行CPU密集型或高内存的数据处理逻辑
#     # ...
# consumer.close()

from fastapi import FastAPI
# import boto3 # AWS SDK for Python

app = FastAPI()
# lambda_client = boto3.client('lambda', region_name='your-region')

@app.post("/process_data_with_lambda")
async def process_data_with_lambda(payload: dict):
    # 调用AWS Lambda函数异步处理数据
    # response = lambda_client.invoke(
    #     FunctionName='your-data-processing-lambda',
    #     InvocationType='Event', # 异步调用
    #     Payload=json.dumps(payload)
    # )
    print(f"数据处理请求已发送到Lambda: {payload}")
    return {"message": "数据处理任务已提交到Lambda"}

# lambda_function.py
import json

def lambda_handler(event, context):
    data_to_process = json.loads(event['body']) # 假设从API Gateway接收POST请求
    print(f"Lambda 正在处理数据: {data_to_process}")
    # 在这里执行CPU密集型或高内存的数据处理逻辑
    # ...
    return {
        'statusCode': 200,
        'body': json.dumps({'message': '数据处理完成'})
    }