Python分布式任务队列Celery处理结果如何入库？

在Celery里处理任务结果入库，最直接的方式是用backend参数。你需要在创建Celery实例时配置一个结果后端，比如用RabbitMQ（RPC）、Redis或者数据库。这里给你一个用Redis做后端，并把结果存到MySQL的完整例子。

首先，装好依赖：

pip install celery redis pymysql

然后是你的tasks.py：

from celery import Celery
import pymysql
from pymysql.cursors import DictCursor
import json
import time

# 配置Celery，用Redis做broker和backend
app = Celery('my_tasks',
             broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/0')

# 一个简单的任务
@app.task
def process_data(data_id):
    # 模拟一些处理
    time.sleep(2)
    result = {'id': data_id, 'status': 'processed', 'value': data_id * 10}
    return result

# 另一个任务，专门用来把结果存到MySQL
@app.task
def save_to_mysql(task_result):
    # 连接MySQL
    connection = pymysql.connect(
        host='localhost',
        user='your_user',
        password='your_password',
        database='celery_results',
        charset='utf8mb4',
        cursorclass=DictCursor
    )
    
    try:
        with connection.cursor() as cursor:
            # 创建表（如果不存在）
            create_table_sql = """
            CREATE TABLE IF NOT EXISTS task_results (
                id INT AUTO_INCREMENT PRIMARY KEY,
                task_id VARCHAR(255) NOT NULL,
                result_data JSON,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            """
            cursor.execute(create_table_sql)
            
            # 插入结果
            # 注意：这里task_result是之前任务的返回值，我们假设它是个字典
            insert_sql = "INSERT INTO task_results (task_id, result_data) VALUES (%s, %s)"
            # 用当前任务的ID作为task_id，实际使用时你可能需要传递原始任务的ID
            cursor.execute(insert_sql, (save_to_mysql.request.id, json.dumps(task_result)))
        
        connection.commit()
    finally:
        connection.close()
    
    return f"Result saved to MySQL for task {save_to_mysql.request.id}"

# 链式调用示例：先处理数据，然后把结果存到MySQL
def process_and_save(data_id):
    # 使用chain：process_data执行完后，结果传给save_to_mysql
    from celery import chain
    result_chain = chain(process_data.s(data_id), save_to_mysql.s())
    return result_chain()

运行worker：

celery -A tasks.app worker --loglevel=info

调用任务：

from tasks import process_and_save

# 这会先执行process_data，然后把结果传给save_to_mysql
async_result = process_and_save(42)
print(f"Task chain started with ID: {async_result.id}")

这个方案的核心是用chain把两个任务串起来，第一个任务处理业务，第二个任务专门负责入库。Redis backend用来存中间状态，MySQL做持久化存储。注意处理好数据库连接和错误重试。

简单说就是用任务链把业务处理和入库分开。

不知道具体业务如何。根据你的描述
1 分理一个写数据库的 task
2 这个 task 只允许在主节点跑

上面可以解决不暴露 mysql 给其他 celery node 这个目标。
另外 celery 是基于消息队列的，这本身就是你写 mysql 的缓冲。