Appearance
高并发下怎么调IP接口才不被打爆?缓存与限流实战
免费的IP查询接口虽然好,但免费版都有个"每IP每分钟60次"的限制。如果你直接把所有用户的请求都怼上去,分分钟触发限流,然后整个功能瘫痪。
这篇文章讲讲高并发场景下,怎么把IP查询接口用得又稳又不浪费。核心就三招:缓存、聚合、降级。
第一招:缓存(最重要)
同一批IP,其实不用每次都查。缓存做得好,接口压力能下降90%以上。
本地缓存(单机够用):
python
from functools import lru_cache
@lru_cache(maxsize=100000)
def query_ip(ip):
...分布式缓存(多机共享):
多台服务器各缓存各的,会重复查询。用 Redis 统一缓存,TTL设个5分钟:
python
import redis, requests
r = redis.Redis(host='127.0.0.1', port=6379)
def query_ip(ip):
# 先查缓存
cached = r.get(f"ip:{ip}")
if cached:
return json.loads(cached)
# 缓存没有,再调接口
data = requests.get(f"https://ip9.com.cn/get?ip={ip}", timeout=3).json()
if data.get("ret") == 200:
r.setex(f"ip:{ip}", 300, json.dumps(data["data"])) # 缓存5分钟
return data["data"]
return None关键细节:缓存过期别用"同时过期"。给TTL加一点随机值,避免所有key在同一秒失效,然后一起打到接口上——那叫缓存雪崩。
python
import random
ttl = 300 + random.randint(0, 60)第二招:聚合请求
如果同一个请求里需要查多个IP(比如一次查一批),别循环一个个请求。IP9 是标准GET,一次查一个。但你可以用"查多个时只查缓存里没有的"策略,把需要真实查询的IP数量压到最低。
还有个思路:热点IP集中查询。有些IP(比如某个机房出口)会被大量请求命中,这种IP查一次缓存起来,收益极大。
第三招:降级与兜底
接口总有不可用的时候(限流、网络抖动)。生产环境必须有降级方案:
python
def safe_query(ip, fallback=None):
try:
data = query_ip(ip)
if data:
return data
except Exception:
pass
# 降级:返回上次结果、返回空、或者用备用接口
return fallback降级三选一:
- 静默:查询失败就不显示,别报错吓到用户;
- 用旧数据:Redis里哪怕缓存过期了,也可以先把旧值返回,后台异步刷新;
- 换备用:准备一个备用查询源,主源挂了自动切换。
一个完整的最小实现
把上面三招拼起来:
python
import redis, requests, random, json
from functools import lru_cache
r = redis.Redis(host='127.0.0.1', port=6379)
@lru_cache(maxsize=10000)
def query_ip(ip):
try:
# 1. 查Redis缓存
cached = r.get(f"ip:{ip}")
if cached:
return json.loads(cached)
# 2. 查接口
data = requests.get(
f"https://ip9.com.cn/get?ip={ip}", timeout=3
).json()
if data.get("ret") == 200:
ttl = 300 + random.randint(0, 60)
r.setex(f"ip:{ip}", ttl, json.dumps(data["data"]))
return data["data"]
except Exception:
pass
return None # 3. 失败返回None,上层降级双缓存(进程内 lru_cache + Redis)的好处是:热点IP连Redis都不用走,性能直接拉满。
一句忠告
免费接口不是给你做压测用的。自己做足了缓存,接口请求量通常能降一个数量级,对大家都好。等哪天你真的并发高到免费版扛不住,该上VIP就上VIP,那本来就是为这个准备的。
IP9 的接口文档:https://www.ip9.com.cn