使用Daphne时Django异步数据库查询卡顿问题及解决

问题背景 #

我开发了一个异步Django应用做数据下载，核心逻辑里用到了异步Django ORM查询（比如async for循环、sync_to_async、.aget()）。为了在客户端实时展示下载进度，我用django-channels和Daphne搭建了WebSocket服务。但发现一个奇怪的问题：没启用Daphne时代码运行完全正常，一旦在settings.py里注册Daphne（启动runserver时用Daphne服务器），异步数据库查询就会卡住，完全无法推进。

环境配置 #

settings.py相关配置：

ASGI_APPLICATION = 'MultiClaw.asgi.application'

INSTALLED_APPS = [
    'daphne',
    ...
]

问题代码 #

启动下载的同步视图 #

@csrf_exempt
def start_grab_thread(request):
    parser_name = request.POST['parser_name']

    settings_model = Settings.objects.filter(user=request.user, parser_name=parser_name)[0]
    parser_module = importlib.import_module(f'parser.Parser.{parser_name}')
    parser_class = getattr(parser_module, parser_name)
    parser_instance: Core = parser_class(settings_model)
    
    asyncio.run(parser_instance.download_products())
    
    if request.method == "POST":       
        return redirect('start')

卡顿的异步查询函数 #

async def download_products(self):
    
    print('downloading products')
    
    collected_product_urls = self.product_url_list

    print(f'{collected_product_urls=}')
            
    product_urls_from_db = set([
        product.source_url async for product in 
        Product.objects
        .filter(pk__in=collected_product_urls)
    ])
    
    print(f'{product_urls_from_db=}')
    print(f'FINALLY GOT IT') 
    return

排查过程 #

我试过各种方法：把查询逻辑封装成单独函数，用database_sync_to_async、async_to_sync装饰；还写了带asyncio.sleep(1)的测试异步函数，发现测试函数能正常运行，只有异步数据库查询会卡住，没有任何报错，就是一直出不来FINALLY GOT IT的输出。

解决方法 #

折腾两天后，我把同步视图改成异步视图就解决了问题，修改细节如下：

原同步视图 #

@csrf_exempt
def start_grab_thread(request):
    parser_name = request.POST['parser_name']

    settings_model = Settings.objects.filter(user=request.user, parser_name=parser_name)[0]
    parser_module = importlib.import_module(f'parser.Parser.{parser_name}')
    parser_class = getattr(parser_module, parser_name)
    parser_instance: Core = parser_class(settings_model)
    
    asyncio.run(parser_instance.download_products())
    
    if request.method == "POST":       
        return redirect('start')

修改后的异步视图 #

@csrf_exempt
async def start_grab_thread(request):
    parser_name = request.POST['parser_name']

    settings_model = await Settings.objects.aget(user=request.user, parser_name=parser_name)
    parser_module = importlib.import_module(f'parser.Parser.{parser_name}')
    parser_class = getattr(parser_module, parser_name)
    parser_instance: Core = parser_class(settings_model)
    
    await parser_instance.download_products()
    
    if request.method == "POST":       
        return redirect('start')

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