如何为Celery Worker分配自定义Python类实例处理队列任务？

核心结论 #

不能直接给Celery Worker分配自定义类实例并跨任务复用，Celery的任务执行模型决定了Worker进程间（甚至同进程的任务间）是隔离的，实例无法在任务间共享。你当前的实现问题在于每个任务都新建Selenium浏览器实例，会造成大量资源占用，而且硬分片的负载不均衡。

当前代码的问题 #

• 每个process_road任务初始化Window，意味着每个任务打开一个独立的浏览器窗口，5个Worker同时运行就会开5个浏览器，数据量越大资源消耗越夸张
• 硬拆成5份的分片逻辑，若数据总量不是5的倍数，最后一批数据量会更大，导致Worker负载不均

正确实现方案：Worker级别复用浏览器实例 #

利用Celery的worker_init信号，在Worker启动时创建一个全局的浏览器实例，让该Worker上的所有任务复用这个实例，避免重复创建资源。

1. 改造Worker初始化与任务逻辑

from celery import signals
from celery_app import celery_app
from your_module import Window

# 存储当前Worker进程的全局浏览器实例
worker_browser = None

# Worker启动时初始化浏览器
@signals.worker_init.connect
def setup_worker_browser(**kwargs):
    global worker_browser
    # 初始化Window类（根据你的实际需求调整初始化参数）
    worker_browser = Window()

# 单条数据处理任务，复用Worker级别的浏览器实例
@celery_app.task
def process_single_road(road_item):
    global worker_browser
    try:
        worker_browser.collect_photo(road_item)
        return True
    except Exception as e:
        # 捕获浏览器崩溃等异常，尝试重新初始化实例
        worker_browser = Window()
        worker_browser.collect_photo(road_item)
        return True

# Worker退出时关闭浏览器，释放资源
@signals.worker_shutdown.connect
def teardown_worker_browser(**kwargs):
    global worker_browser
    if worker_browser:
        worker_browser.close_browser()  # 假设Window类有关闭浏览器的方法

2. 优化任务提交逻辑

放弃硬分片，改为提交单条数据的任务，让Celery自动做负载均衡：

from flask import jsonify

def process(road_data):
    # 逐个提交单条数据任务
    task_ids = [process_single_road.delay(item).id for item in road_data]
    
    # 若想批量提交，可用Celery的group提高效率
    # from celery import group
    # job = group(process_single_road.s(item) for item in road_data)
    # result = job.apply_async()
    # task_ids = result.task_ids
    
    return jsonify({'task_ids': task_ids}), 202

关键说明 #

• Celery任务是序列化后通过Broker传递的，自定义类实例无法直接序列化传递到Worker，即使强行序列化，传递后也是重新创建的实例，达不到复用目的
• 每个Worker进程对应一个浏览器实例，进程内的所有任务复用该实例，既节省资源，又避免多线程操作浏览器的冲突（Celery默认用进程池，无需考虑线程安全）
• 务必处理浏览器崩溃的异常，避免单个任务失败导致整个Worker进程无法继续处理后续任务

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