多服务器IP白名单配置(使用redis stream实现)

应用背景

现在我有一个管理平台，可以通过代理连接到内网网站，但是这个代理服务器没有设置密码，所以需要IP白名单让指定用户才可以使用代理。

添加白名单流程图

流程描述：

1. 登录管理平台成功后，管理平台的后台将这个登录的IP地址添加到redis，并设置过期时间为24小时
2. redis通过stream将这个IP分发给每台代理服务器
3. 代理服务器通过redis客户端接收到消息后，更新白名单
4. 如果IP地址过期了，管理平台也会发送消息通知每台代理服务器，将过期的IP地址从每台代理服务器上移除

使用Python实现

• 管理平台端

  1. redis客户端基本配置

     import asyncio
     from typing import AsyncIterator
     import aioredis
     
     from src.redis_client.redis_config import redis_config
     from src.redis_client.log import log
     
     class AsyncRedisClient:
         def __init__(self):
             self.db = redis_config.db
             self.password = redis_config.password
             self.port = redis_config.port
             self.host = redis_config.host
             self.connect_pool = aioredis.ConnectionPool.from_url(
                 f"redis://{
              self.host}:{
              self.port}",
                 db=self.db,
                 password=self.password,
                 decode_responses=True,
             )
             self._stream_key = "ip_whitelist_stream"  # Stream消息队列键名
     
         async def connect(self):
             self.client = aioredis.Redis(connection_pool=self.connect_pool)
     
         async def close(self):
             await self.client.close()
     

  2. 添加IP到白名单，并通过Stream通知所有客户端

         async def add_ip_to_whitelist(
             self, ip: str, expire_seconds: int = 3600 * 24
         ) -> bool:
             """
             添加IP到白名单，并通过Stream通知所有客户端
             :param ip: IP地址
             :param expire_seconds: 过期时间（秒）
             :return: 是否添加成功
             """
             # 1. 使用Set存储IP（确保唯一性）
             added = await self.client.sadd("ip_whitelist", ip)
             if not added:
                 return False  # IP已存在
     
             # 2. 设置TTL（通过单独的键实现精确过期控制）
             await self.client.setex(f"ip_whitelist_ttl:{
              ip}", expire_seconds, "active")
     
             # 3. 发送Stream消息通知其他客户端（包含操作类型和过期时间）
             await self.client.xadd(
                 self._stream_key,
                 {
            "ip": ip, "action": "add", "expire": str(expire_seconds)},
                 maxlen=1000,  # 限制Stream长度防止内存溢出
             )
             log.info(f"发布消息到Stream: {
              ip}")
             return True
     

  3. 移除代理IP并发送通知

         async def remove_ip_from_whitelist(self, ip: str) -> bool:
             """移除代理IP并发送通知"""
             removed = await self.client.srem("ip_whitelist", ip)
             if removed:
                 await self.client.delete(f"ip_whitelist_ttl:{
              ip}")
                 await self.client.xadd(
                     self._stream_key, {
            "ip"