Django中使用Async Redis Client实现分布式信号量遇事件循环错误

你遇到的「got Future attached to a different loop」错误，核心原因是Redis连接池与事件循环绑定不匹配：

• 在settings.py中提前初始化的共享连接池，会绑定到Django启动时的事件循环；
• 而Django的ASGI服务器（如Uvicorn）会为每个请求（或请求组）创建新的事件循环，当信号量类复用旧循环的连接池时，连接池内的连接归属与当前请求循环冲突，从而触发报错。

方法1：动态创建适配当前循环的连接池 #

不在settings.py提前初始化连接池，而是在信号量类中根据当前事件循环动态创建或获取对应连接池：

import asyncio
from redis.asyncio import ConnectionPool, Redis

# 用字典存储不同事件循环对应的连接池
pool_map = {}

class DistributedSemaphore:
    def __init__(self, key, max_connections):
        self.key = key
        self.max_connections = max_connections
        self.loop = asyncio.get_running_loop()
        # 确保当前循环有专属连接池
        if self.loop not in pool_map:
            pool_map[self.loop] = ConnectionPool.from_url("redis://localhost:6379")
        self.pool = pool_map[self.loop]
        self.redis = Redis(connection_pool=self.pool)

    # 信号量获取逻辑（示例）
    async def acquire(self):
        while await self.redis.incr(self.key) > self.max_connections:
            await self.redis.decr(self.key)
            await asyncio.sleep(0.1)
        return True

    # 信号量释放逻辑（示例）
    async def release(self):
        await self.redis.decr(self.key)

每个事件循环都有独立的连接池，彻底避免跨循环复用连接的问题。

方法2：封装适配ASGI的单例Redis客户端 #

通过单例模式动态生成客户端，确保每次获取的客户端绑定当前事件循环：

import asyncio
from redis.asyncio import Redis

class AsyncRedisClient:
    _instance = None

    @classmethod
    async def get_client(cls):
        current_loop = asyncio.get_running_loop()
        # 若实例不存在或绑定的循环不匹配，重建客户端
        if cls._instance is None or cls._instance.connection_pool._loop != current_loop:
            cls._instance = Redis.from_url("redis://localhost:6379")
        return cls._instance

# 信号量类中使用
class DistributedSemaphore:
    def __init__(self, key, max_connections):
        self.key = key
        self.max_connections = max_connections
        self.redis = None

    async def acquire(self):
        if self.redis is None:
            self.redis = await AsyncRedisClient.get_client()
        # 后续信号量逻辑...

方法3：调整ASGI服务器循环策略（辅助优化） #

若使用Uvicorn，可通过--loop参数指定统一循环策略，减少循环创建的差异：

uvicorn main:app --loop uvloop --workers 4

但此方法仅为辅助优化，无法完全避免动态循环的问题，建议配合前两种方法使用。

• 应用关闭时清理连接池：可在Django的ready()信号中遍历pool_map，调用pool.disconnect()关闭所有连接池；
• 保证信号量操作原子性：建议用Redis Lua脚本实现acquire和release，避免竞态条件，示例Lua脚本：

-- acquire原子脚本
if redis.call('incr', KEYS[1]) <= ARGV[1] then
    return 1
else
    redis.call('decr', KEYS[1])
    return 0
end

在Python中通过self.redis.eval()执行脚本即可。

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