应用背景

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

添加白名单流程图

流程描述：

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

使用Python实现

• 管理平台端

  1. redis客户端基本配置

     import asyncio
     from typing import AsyncIterator
     import aioredisfrom src.redis_client.redis_config import redis_config
     from src.redis_client.log import logclass AsyncRedisClient:def __init__(self):self.db = redis_config.dbself.password = redis_config.passwordself.port = redis_config.portself.host = redis_config.hostself.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"