运行异步Celery任务时Django测试冻结问题求助

问题背景 #

在Django单元测试中验证Celery任务upload_csv通过CSV创建对象时，调用AsyncResult.get()会导致测试冻结。但Celery日志显示任务已执行成功，数据库也有数据写入，生产环境功能正常，仅测试环境出现该问题。

已尝试的无效方案：

• 使用@patch模拟任务，但无法验证真实数据库对象创建
• 切换为SQLite数据库，测试仍冻结
• 创建简单临时任务，同样出现冻结

相关代码 #

stations/test.py #

class StationsTests(TestCase):
    
    @classmethod
    def setUpTestData(cls):
        super(StationsTests, cls).setUpTestData()
        dirname = os.path.abspath(os.path.join(os.path.dirname( __file__ ), '..', 'csvimport'))

        cls.csv_data_type_station = 'station'
        cls.upload_type = 'safe_create'
        cls.file_station = os.path.join(dirname, 'CSVFiles/Station_test_csv.csv')


    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                    CELERY_ALWAYS_EAGER=True)
    def test_station_geoJSON_data(self):
        self.result_upload_csv = upload_csv.delay(
            self.file_station, self.csv_data_type_station, self.upload_type).get(timeout=10)

        self.assertEqual(self.result_upload_csv, '10 stations uploaded successfully')
        self.assertEqual(len(Station.objects.all()), 10)

settings.py #

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql_psycopg2',
        'NAME': 'db_hcbapp',
        'USER': 'postgres',
        'PASSWORD': 'postgres',
        'HOST': 'localhost',
        'PORT': '5432'
    },
}
.....

CELERY_BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_IMPORTS = ('csvimport.tasks')
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_RESULT_EXTENDED = True

celery.py #

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')
app = Celery('myapp')
app.conf.enable_utc = False
app.config_from_object(settings, namespace='CELERY')

app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

tasks.py #

@shared_task(bind=True)
def upload_csv(self, file_path, csv_data_type, upload_type):

    progress_recorder = ProgressRecorder(self)
    csv_data = []   # Temporary list for storing the CSV-rows
    if not bool(upload_type == "safe_create") | bool(upload_type == "chunk_create"):
        return "The upload type needs to be either Safe Create or Chunk Create"

    with open(file_path, 'r') as file:
        reader = csv.DictReader(file)

        if csv_data_type == 'station':
            pattern = "FID,ID,Nimi,Namn,Name,Osoite,Adress,Kaupunki,Stad,Operaattor,Kapasiteet,x,y"
            pattern = pattern.replace("(", "\(").replace(")", "\)")
            if not re.search(pattern, ",".join(reader.fieldnames), re.IGNORECASE):
                raise Exception(
                    "CSV invalid as it does not contain the correct fieldnames '['FID','ID','Nimi','Namn','Name','Osoite','Adress','Kaupunki','Stad','Operaattor','Kapasiteet','x','y']' for a Station upload.")

            # Fields in the CSV (and the index number): [0]FID,[1]ID,[2]Nimi,[3]Namn,[4]Name,[5]Osoite,[6]Address,[7]Kaupunki,[8]Stad,[9]Operaattor,[10]Kapasiteet,[11]x,[12]y
            # Fieldnames are reassigned just in case so the object data can be gathered from correctly named row key-names.
            reader.fieldnames = ['FID', 'ID', 'Nimi', 'Namn', 'Name', 'Osoite',
                                 'Adress', 'Kaupunki', 'Stad', 'Operaattor', 'Kapasiteet', 'x', 'y']

        print("Reading the csv")
        for row in reader:
            csv_data.append(row)
        file.seek(0)

    # Creating objects ->>
    rowcount = len(csv_data)
    print(f"CSV read complete with {rowcount} rows. Creating objects...")

    try:
        object_list = []    # Temporary list for the objects about to be created in bulk
        if csv_data_type == 'station':

            print("Uploading a Station")
            for i, row in enumerate(csv_data):

                # Check empty rows
                if any(value for value in row.values()):
                    try:
                        csv_data = Station(
                            station_id=row['ID'],
                            fid=row['FID'],
                            name_fin=row['Nimi'],
                            name_swe=row['Namn'],
                            name_eng=row['Name'],
                            address_fin=row['Osoite'],
                            address_swe=row['Adress'],
                            city_fin=row['Kaupunki'],
                            city_swe=row['Stad'],
                            operator=row['Operaattor'],
                            capacity=int(row['Kapasiteet']),
                            geo_pos_x=Decimal(row['x']),
                            geo_pos_y=Decimal(row['y'])
                        )

                        # If station is found in the db, skip to the next iteration
                        if Station.objects.filter(station_id=csv_data.station_id).exists():
                            print("Station already exists")
                            continue

                        # Add the object to the list
                        object_list.append(csv_data)

                        # Chunk create checks if there are 100 Journey-objects, if so, creates them and clears the temporary list
                        if upload_type == 'chunk_create':
                            if len(object_list) >= 100:
                                Station.objects.bulk_create(object_list)
                                # Clear the list
                                object_list = []

                        print(
                            f"Row {i+1} : Station {row['ID']}.{row['Nimi']} created!")

                    except ValueError as e:
                        print(f"Row {i+1} : Value error in row {i+1}! ", e)

                    except Exception as e:
                        print(f"Row {i+1} : Exception: {e}")
                        continue
                else:
                    # In case a csv-row is empty
                    print(f"Row {i+1} : Field empty")

                # Updates the progress after every iteration and passes the current state (in percentages) to the front end
                progress_recorder.set_progress(
                    i + 1, rowcount, f'{round(((i+1) / rowcount) * 100),2}%')
            # If there are leftover objects in the list after bulk_create or safe_create has finished creating all objects, add them to the db.
            if object_list:
                Station.objects.bulk_create(object_list)
            return f"{rowcount} stations uploaded successfully"

        # Just in case neither Journey or Station upload_type is chosen
        else:
            print("Select a CSV containing proper values of either Stations or Journeys")

    except Exception as e:
        print(f"Exception, something went wrong: {e}")
        return f"Something went wrong: {e}"

问题原因分析 #

1. Django测试事务与Celery结果后端冲突：Django的TestCase默认开启事务，测试结束后回滚。而你使用CELERY_RESULT_BACKEND = 'django-db'，任务执行后将结果写入数据库，但该写操作处于测试事务的子事务中，get()方法尝试读取结果时，因主事务未提交导致死锁。
2. Eager模式下的结果存储逻辑：设置CELERY_ALWAYS_EAGER=True时，任务在当前线程执行，但django-db结果后端仍会尝试将结果写入数据库，测试事务的隔离性导致结果无法被读取，get()会一直等待结果更新。
3. ProgressRecorder的影响：任务中使用的ProgressRecorder会频繁更新任务状态到数据库，同样受测试事务的影响，状态更新无法被主线程读取，导致get()阻塞。

解决思路 #

方案1：修改测试类的事务行为 #

使用TransactionTestCase替代TestCase，它不会自动开启事务，测试结束后直接删除数据，避免事务隔离导致的结果读取问题：

from django.test import TransactionTestCase

class StationsTests(TransactionTestCase):
    # 其余代码保持不变

方案2：测试环境临时替换Celery结果后端 #

在测试中覆盖CELERY_RESULT_BACKEND为内存后端，避免数据库写入的事务问题：

@override_settings(
    CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
    CELERY_ALWAYS_EAGER=True,
    CELERY_RESULT_BACKEND='cache+memory://'
)
def test_station_geoJSON_data(self):
    # 测试代码保持不变

方案3：直接调用任务而非使用delay().get() #

在Eager模式下，直接调用任务函数即可，无需通过Celery的异步接口，避免结果后端的读写问题：

def test_station_geoJSON_data(self):
    self.result_upload_csv = upload_csv(
        self.file_station, self.csv_data_type_station, self.upload_type
    )
    # 断言代码保持不变

方案4：禁用任务的进度记录（临时排查） #

如果ProgressRecorder是阻塞诱因，可在测试中临时禁用它，或修改任务逻辑在测试环境跳过进度更新：

# 在tasks.py中添加测试环境判断
import os
if not os.environ.get('TESTING'):
    progress_recorder.set_progress(...)

然后在测试运行时设置环境变量TESTING=1。

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