运营商三要素实名认证：按次计费下的缓存策略
idcard-phone-auth-cache-cost-1389 # 运营商三要素实名认证：按次计费下的缓存策略
> **接口**：运营商三要素 - 运营商手机号实名认证 · **apiCode**：1389 · **接入点**：1389-1
> **请求方式**：POST · **返回格式**：JSON · **计费**：按次（ret_code=0 时扣费，具体档位以官方说明为准）
> **适用人群**：已接入用户、中高级开发者
> **阅读时间**：8 分钟
> **最后实测核对**：2026-09-07
---
## 核心要点
- 同一组（姓名+身份证+手机号）的认证结果具有**时间稳定性**：只要用户信息不变，认证结果就不会变。
- 合理缓存可以将重复调用的成本降为零，同时降低接口响应时间。
- 缓存 key 必须包含全部三个字段，避免不同用户的数据混淆。
---
## 为什么需要缓存
### 成本测算示例
假设你的业务场景：
- 日均注册用户 10,000 人
- 其中 30% 会在 24 小时内重复提交注册表单（网络卡顿、误操作等）
- 每次调用成本 ¥0.05（示例，实际以官方档位为准）
**无缓存**：10,000 × 30% × ¥0.05 = ¥150/天，月度 ¥4,500
**有缓存**：重复请求走缓存，成本降低 30%，月度节省 ¥1,350
对于日均 10 万用户的平台，月度节省可达 ¥13,500。规模越大，缓存价值越明显。
---
## 缓存策略设计
### 缓存什么
| 认证结果 | 缓存有效期 | 理由 |
|---------|-----------|------|
| code=0（认证成功） | 30 天 | 用户身份信息短期内不会变更 |
| code=1（认证失败） | 24 小时 | 可能是输入错误，次日可重试 |
| code=2（无记录） | 7 天 | 用户可能已完成实名，一周后重试 |
| code=11/12/13（参数错误） | 不缓存 | 参数错误不是业务状态，缓存无意义 |
| code=21/22（渠道异常） | 不缓存 | 渠道状态动态变化，缓存可能导致误判 |
### 缓存 Key 设计
```
key 格式：auth:{md5(name + idCard + phone)}
```
使用 MD5 哈希的原因：
1. 保护用户隐私：不直接存储明文身份证号和姓名。
2. 固定长度：MD5 输出 32 位十六进制字符串，适合作为 Redis key。
3. 碰撞概率低：对于 10 亿级用户，碰撞概率可忽略。
**注意**：MD5 仅用于 key 生成，不用于密码存储（密码应用 bcrypt/argon2）。
### 缓存 Value 设计
```json
{
"code": 0,
"msg": "认证成功",
"belongArea": {
"prov": "山西",
"city": "临汾市",
"name": "移动神州行卡",
"num": 1362068,
"provCode": "140000",
"type": 1
},
"cached_at": "2026-09-07T15:30:00+08:00",
"ttl": 2592000
}
```
---
## 生产级 Redis 缓存代码（Python）
```python
import hashlib
import json
import redis
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
class AuthCache:
"""运营商三要素认证缓存管理器"""
def __init__(self, redis_url: str = "redis://localhost:6379/0"):
self.redis = redis.from_url(redis_url)
# 各状态的缓存 TTL（秒）
self.ttl_map = {
0: 30 * 24 * 3600, # 认证成功：30 天
1: 24 * 3600, # 认证失败：24 小时
2: 7 * 24 * 3600, # 无记录：7 天
# 11/12/13/21/22 不缓存
}
def _make_key(self, name: str, id_card: str, phone: str) -> str:
"""生成缓存 key"""
raw = f"{name}{id_card}{phone}"
hash_val = hashlib.md5(raw.encode()).hexdigest()
return f"auth:{hash_val}"
def get(self, name: str, id_card: str, phone: str) -> Optional[Dict[str, Any]]:
"""查询缓存，命中返回认证结果，未命中返回 None"""
key = self._make_key(name, id_card, phone)
value = self.redis.get(key)
if value:
return json.loads(value)
return None
def set(self, name: str, id_card: str, phone: str,
code: int, msg: str, belong_area: Optional[Dict] = None) -> bool:
"""写入缓存，仅缓存 code=0/1/2 的结果"""
if code not in self.ttl_map:
return False # 参数错误或渠道异常，不缓存
key = self._make_key(name, id_card, phone)
ttl = self.ttl_map[code]
cache_data = {
"code": code,
"msg": msg,
"belongArea": belong_area,
"cached_at": datetime.now().isoformat(),
"ttl": ttl
}
self.redis.setex(key, ttl, json.dumps(cache_data, ensure_ascii=False))
return True
def invalidate(self, name: str, id_card: str, phone: str) -> bool:
"""手动清除缓存（用户主动修改信息时调用）"""
key = self._make_key(name, id_card, phone)
return bool(self.redis.delete(key))
```
### 使用示例
```python
# 初始化缓存管理器
cache = AuthCache(redis_url="redis://your-redis-server:6379/0")
# 业务逻辑
def check_authentication(name: str, id_card: str, phone: str) -> Dict:
# 1. 先查缓存
cached = cache.get(name, id_card, phone)
if cached:
return {
"source": "cache",
"code": cached["code"],
"msg": cached["msg"],
"belong_area": cached.get("belongArea")
}
# 2. 缓存未命中，调用接口
result = call_showapi_1389(name, id_card, phone)
code = result["code"]
# 3. 写入缓存（仅成功/失败/无记录）
cache.set(
name=name,
id_card=id_card,
phone=phone,
code=code,
msg=result["msg"],
belong_area=result.get("belongArea")
)
return {
"source": "api",
**result
}
```
---
## 缓存击穿防护
### 问题描述
大量并发请求同时查询同一个未缓存的 key，导致所有请求同时打到接口，造成瞬时压力。
### 解决方案：分布式锁
```python
import threading
# 简单的内存锁（单机场景）
_locks = {}
def check_with_cache_and_lock(name: str, id_card: str, phone: str) -> Dict:
# 1. 查缓存
cached = cache.get(name, id_card, phone)
if cached:
return {"source": "cache", **cached}
# 2. 分布式锁防止击穿
lock_key = f"lock:auth:{hashlib.md5((name + id_card + phone).encode()).hexdigest()}"
# 尝试获取锁（Redis SET NX EX）
acquired = cache.redis.set(lock_key, "1", nx=True, ex=5)
if not acquired:
# 等其他请求缓存后重试
import time
time.sleep(0.1)
cached = cache.get(name, id_card, phone)
if cached:
return {"source": "cache", **cached}
# 锁未获取且缓存仍未命中，降级为直接调用
result = call_showapi_1389(name, id_card, phone)
cache.set(name, id_card, phone, result["code"], result["msg"], result.get("belongArea"))
return {"source": "api", **result}
try:
# 双重检查（获取锁后再次查缓存）
cached = cache.get(name, id_card, phone)
if cached:
return {"source": "cache", **cached}
# 调用接口
result = call_showapi_1389(name, id_card, phone)
cache.set(name, id_card, phone, result["code"], result["msg"], result.get("belongArea"))
return {"source": "api", **result}
finally:
cache.redis.delete(lock_key)
```
---
## 缓存更新策略
### 用户主动修改信息时清除缓存
```python
def on_user_info_changed(user_id: int, new_name: str, new_id_card: str, new_phone: str):
"""用户修改实名信息时清除旧缓存"""
# 查询该用户的历史认证记录
old_records = db.query_user_auth_records(user_id)
for record in old_records:
cache.invalidate(record.name, record.id_card, record.phone)
# 写入新信息的缓存（预填充）
cache.set(new_name, new_id_card, new_phone, None, None) # 占位，待实际认证后更新
```
### 定期清理过期缓存
Redis 的 `EXPIRE` 会自动清理过期 key，无需手动维护。但如果需要统计缓存命中率，可以定期导出 metrics。
---
## FAQ
**Q1：缓存命中率一般能达到多少？**
取决于业务场景。注册场景的重复提交率通常在 10%-30%，意味着缓存命中率相近。风控场景（用户多次触发验证）可能高达 50% 以上。
**Q2：用户修改了手机号，旧缓存会影响新号码吗？**
不会。缓存 key 基于"姓名+身份证+手机号"三元组哈希，手机号不同则 key 不同，互不影响。但建议用户修改信息时主动清除旧缓存（见上方代码）。
**Q3：Redis 挂了怎么办？**
降级为直接调用接口。缓存是性能优化手段，不应成为功能依赖。代码中应有 fallback 逻辑。
**Q4：缓存的 30 天有效期合理吗？**
对于实名认证场景，用户身份信息短期内变更概率极低，30 天是保守估计。如果业务有特殊要求（如金融场景需每日复核），可缩短至 24 小时。
**Q5：多实例部署时缓存如何共享？**
使用集中式 Redis 即可。所有应用实例连接同一个 Redis 集群，缓存天然共享。注意 Redis 集群的高可用配置（主从复制 + Sentinel 或 Cluster 模式）。
---
## 下一步阅读
- [完整错误码对照表](https://www.showapi.com/guides/idcard-phone-auth-response-codes-1389) —— code=21/22 时不要缓存
- [注册实名校验集成方案](https://www.showapi.com/guides/idcard-phone-auth-scenario-integration-1389) —— 业务层如何配合缓存使用
- [三要素 vs 二要素 vs 四要素方案对比](https://www.showapi.com/guides/idcard-phone-auth-comparison-1389) —— 选型决策参考
---
**- 本系列共 6 篇**：查看[运营商三要素实名认证指南总目录](https://www.showapi.com/guides/idcard-phone-auth-guides-1389)
