使用pytest-xdist运行Celery测试时出现死锁问题求助

问题现象 #

• 单进程运行测试正常：

  pytest --disable-warnings --verbose -s test_celery_chords.py
  

• 使用xdist多进程（-n 2）时，进程挂起，偶发Chord回调ValueError，只能强制终止：

  pytest --disable-warnings --verbose -n 2 -s test_celery_chords.py
  

核心原因 #

1. 多进程资源冲突：xdist的每个测试进程都独立启动Celery worker，导致Broker/结果后端出现竞争，Chord的组任务完成信号无法被正确传递给回调任务。
2. Fixture作用域不当：默认celery_worker是function级作用域，每个测试用例启动新worker，多进程下重复创建的worker会干扰Chord的状态跟踪。
3. 结果后端多进程兼容性：如果使用SQLite作为结果后端，多进程写操作会导致锁等待，直接引发挂起。

解决方案 #

1. 调整Fixture作用域为Session级 #

修改conftest.py，将Celery相关fixture的作用域改为session，确保整个测试会话只启动一个Celery worker，避免多进程重复创建：

import pytest
from celery.contrib.testing.worker import start_worker

@pytest.fixture(scope="session")
def celery_app(request):
    from celery import Celery
    app = Celery("test_app")
    app.config_from_object("django.conf:settings", namespace="CELERY")
    return app

@pytest.fixture(scope="session")
def celery_worker(celery_app):
    worker = start_worker(celery_app, concurrency=2)
    worker.start()
    yield worker
    worker.stop()

@pytest.fixture(scope="session")
def celery_not_eager(settings):
    settings.CELERY_TASK_ALWAYS_EAGER = False
    settings.CELERY_TASK_EAGER_PROPAGATES = False

2. 配置兼容多进程的Broker/结果后端 #

确保Celery使用支持多进程并发的Broker和结果后端，比如Redis（优先推荐），避免用SQLite这类单进程安全的存储：
在Django settings中修改：

CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"

如果必须用数据库，使用PostgreSQL/MySQL这类支持多进程写的数据库，同时确保transactional_db fixture的隔离性正确。

3. 优化测试用例逻辑 #

• 移除测试用例中重复的celery_app.config_from_object调用，因为session级的celery_app已经完成配置。
• 显式设置任务超时，避免get()无限阻塞：

def test_chords(transactional_db, celery_app, celery_worker, celery_not_eager):
    task = do_average.delay()
    # 设置超时时间，避免无限挂起
    results = task.get(timeout=10)
    assert task.state == "SUCCESS"
    assert len(results[0][1][1]) == 10

def test_chord_differently(transactional_db, celery_app, celery_worker, celery_not_eager):
    task = do_average.delay()
    results = task.get(timeout=10)
    assert task.state == "SUCCESS"
    assert len(results[0][1][1]) == 10

4. 升级pytest-celery版本 #

当前使用的pytest-celery==0.0.0是开发早期版本，存在多进程兼容问题，建议升级到最新稳定版：

pip install --upgrade pytest-celery

如果没有稳定版，直接用Celery自带的celery.contrib.testing模块管理worker即可。

5. 配置xdist进程组隔离（可选） #

如果上述方案无效，可在pytest.ini中添加配置，确保session级fixture只初始化一次：

[pytest]
xdist_group_scope = session

验证 #

修改后重新运行多进程测试：

pytest --disable-warnings --verbose -n 2 -s test_celery_chords.py

确认测试正常完成，无挂起或报错。

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