使用Django的@shared_task时遭遇Broken pipe及MySQL连接异常问题

一、Broken Pipe错误修复 #

问题根源 #

你当前直接调用nonFollowers_users()函数，并没有触发Celery异步任务，而是在请求线程中同步执行耗时任务。当用户刷新/关闭页面时，客户端主动断开连接，但服务器仍在向已关闭的连接写入响应，从而抛出BrokenPipeError。另外，你的异常捕获范围仅包裹return语句，无法覆盖响应发送过程中的连接断开情况。

解决步骤 #

1. 改为异步调用任务：用Celery的delay()方法将任务发送到队列，让请求立刻返回，避免阻塞：

   # 替换直接调用为异步调用，传递account id而非对象（避免序列化问题）
   nonFollowers_users.delay(id, count, sleep)
   

2. 正确捕获连接异常：将整个响应返回逻辑包裹在异常捕获块中，确保客户端断开时能正确返回状态码：

   for id in accounts:
       nonFollowers_users.delay(id, count, sleep)
   
   try:
       return JsonResponse({'success': True})
   except (ConnectionResetError, BrokenPipeError):
       return HttpResponse(status=499)
   

二、MySQL server had gone away错误修复 #

问题根源 #

Celery Worker的数据库连接可能因闲置过久被MySQL服务器关闭，或者你直接传递的Account对象持有已失效的数据库连接。此外，Django的数据库连接池不会自动回收失效连接，导致任务执行时使用无效连接。

解决步骤 #

1. 任务内重置数据库连接：在任务开头调用close_old_connections()，强制关闭失效连接并获取新连接：

   from django.db import close_old_connections
   from celery import shared_task
   
   @shared_task
   def nonFollowers_users(account_id, count, sleep):
       # 关闭旧的失效连接，确保使用新连接
       close_old_connections()
       # 通过id重新获取Account对象，而非直接传递序列化后的对象
       account = Account.objects.get(id=account_id)
       
       # 你的任务逻辑...
       
       # 更新Account对象
       account.save()
   

2. 传递ID而非对象：避免直接传递Account对象给Celery任务，因为序列化后的对象可能携带失效的连接上下文，改用ID在任务内重新查询对象，保证连接有效性。
3. 优化数据库配置：
   • 在Django的settings.py中设置CONN_MAX_AGE，控制连接池的连接生命周期（例如CONN_MAX_AGE=300，即5分钟）
   • 调整MySQL的wait_timeout参数，延长连接闲置超时时间（需根据服务器配置调整）

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