如何在Django中让各调用始终使用同一MySQL Connection ID

核心原因 #

1. Django的ConnectionProxy是连接代理对象，并非真实数据库连接。它会在每次请求时从数据库连接池中获取可用的真实MySQL连接，所以即使代理对象内存地址不变，实际拿到的连接实例可能不同，导致connection_id变化。
2. MySQL的会话级临时表仅对创建它的数据库连接会话可见，当该连接被放回池并重置、或直接关闭时，临时表会自动销毁，其他连接（包括后续请求的新连接）无法访问。

解决方案 #

方案1：改用全局临时表（推荐，适合跨连接访问临时数据）

MySQL 8.0.16及以上版本支持全局临时表，这类表会在所有引用它的会话断开后才被删除，不同连接均可访问：

• 创建表时添加GLOBAL关键字：

CREATE TEMPORARY TABLE IF NOT EXISTS my_global_temp_table (
    id INT PRIMARY KEY,
    content VARCHAR(255)
) GLOBAL;

• 注意：为避免不同业务场景的数据冲突，建议添加唯一标识字段（如business_key）来区分不同请求的临时数据。

方案2：强制复用连接（不推荐，仅适合低并发场景）

如果必须使用会话临时表，可手动强制Django在指定请求链中复用同一个连接：

from django.db import connection

def func_a(request):
    # 手动建立连接并保持打开状态
    connection.connect()
    with connection.cursor() as cursor:
        # 创建会话临时表
        cursor.execute("CREATE TEMPORARY TABLE my_temp_table (id INT);")
        cursor.execute("INSERT INTO my_temp_table VALUES (1);")
    # 不主动关闭连接，留给后续请求复用
    return HttpResponse("临时表创建完成")

def func_b(request):
    # 复用当前连接查询临时表
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM my_temp_table;")
        result = cursor.fetchall()
    return HttpResponse(f"查询结果：{result}")

• 风险：这种方式会破坏连接池的复用机制，导致连接长期占用，高并发下容易引发连接耗尽问题。

方案3：用缓存/持久化存储替代临时表（最稳妥）

彻底避开数据库连接的限制，改用缓存或普通表存储临时数据：

缓存方案（推荐，性能最优）

用Django缓存框架（如Redis、Memcached）存储临时数据：

from django.core.cache import cache

def func_a(request):
    temp_data = {"id": 1, "content": "test_data"}
    # 设置5分钟过期时间
    cache.set("temp_data_xxx", temp_data, timeout=300)
    return HttpResponse("临时数据存入缓存")

def func_b(request):
    temp_data = cache.get("temp_data_xxx")
    return HttpResponse(f"缓存数据：{temp_data}")

普通表方案（适合需要持久化的临时数据）

创建带过期时间的普通表，定时清理过期数据：

# models.py
from django.db import models
from django.utils import timezone

class TempData(models.Model):
    data = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)
    expires_at = models.DateTimeField()

    @classmethod
    def clear_expired(cls):
        cls.objects.filter(expires_at__lt=timezone.now()).delete()

# 函数A
def func_a(request):
    TempData.clear_expired()
    temp_data = {"id": 1, "value": "demo"}
    TempData.objects.create(
        data=temp_data,
        expires_at=timezone.now() + timezone.timedelta(minutes=5)
    )
    return HttpResponse("临时数据存入普通表")

# 函数B
def func_b(request):
    TempData.clear_expired()
    temp_item = TempData.objects.first()
    return HttpResponse(f"查询结果：{temp_item.data if temp_item else '无数据'}")

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