應用背景

現在我有一個管理平臺，可以通過代理連接到內網網站，但是這個代理服務器沒有設置密碼，所以需要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"