Django异步代码中MySQL连接断开后如何实现重连？

我有一个Django项目，MySQL数据库部署在独立主机上，通过TCP连接。编写了如下测试脚本：

#!/usr/bin/env python

from asyncio import sleep, run

from django import setup as django_setup
django_setup()

from django.db import connections
from django.contrib.auth.models import User
from django.db import OperationalError, close_old_connections

async def test():
    while True:
        try:
            objs=User.objects.all()
            print(await objs.acount())
        except KeyboardInterrupt:
            break
        except OperationalError as e: # Need to reconnect
            print(f'main(): OperationalError: {e}')
            # (1053, 'Server shutdown in progress')
            # (2013, 'Lost connection to MySQL server during query')
            # (2006, 'Server has gone away')
            code=e.args[0]
            if code in {1053, 2006, 2013, 2026}: # Only when it's restarting, and once
                #await sync_to_async(conn.connect)()
                close_old_connections()
                print('After reconnect')
        except Exception as e:
            print(f'in _round() {type(e)}')
        else:
            pass
            #breakpoint()

        print('Sleeping')
        await sleep(5)

run(test())

使用tcpkill命令中断连接：

tcpkill 'dst port 3306'

当前Django无法自动重连，调用close_old_connections()也无效，报错输出如下：

3
Sleeping
3
Sleeping
main(): OperationalError: (2013, 'Lost connection to MySQL server during query')
After reconnect
Sleeping
main(): OperationalError: (2013, 'Lost connection to MySQL server during query')
After reconnect
Sleeping
main(): OperationalError: (2013, 'Lost connection to MySQL server during query')
After reconnect
Sleeping

查看代码发现，连接对象仍存在，但生成的游标不可用。尝试调用connection.connect()，但因代码是异步环境无法正常工作。请问如何让Django在异步代码中实现MySQL连接断开后的重连？

• 用sync_to_async包装同步连接操作
  Django的数据库连接方法是同步的，在异步环境中必须通过asgiref.sync.sync_to_async包装后调用。修改异常处理逻辑：

from asgiref.sync import sync_to_async

# ... 保留原有代码 ...

except OperationalError as e:
    print(f'main(): OperationalError: {e}')
    code = e.args[0]
    if code in {1053, 2006, 2013, 2026}:
        # 先关闭失效的旧连接
        close_old_connections()
        # 主动获取默认连接并重建（包装为异步操作）
        conn = connections['default']
        await sync_to_async(conn.connect)()
        print('Reconnected successfully')

• 主动校验连接状态
  在每次执行查询前，主动校验连接有效性，避免使用失效连接。可以在查询逻辑前添加：

from asgiref.sync import sync_to_async

async def test():
    while True:
        try:
            # 校验并确保连接有效，无效则自动重建
            conn = connections['default']
            await sync_to_async(conn.ensure_connection)()
            
            objs = User.objects.all()
            print(await objs.count())  # 修正原代码笔误：acount → count
        # ... 保留原有异常处理逻辑 ...

ensure_connection()会自动检查连接状态，若已断开则重新建立，同样需要用sync_to_async适配异步环境。

• 修正代码笔误
  原测试脚本中objs.acount()是拼写错误，需改为objs.count()，否则即使连接正常也会触发异常。

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