一、系统架构设计

电话号码查询系统需满足高并发、低延迟的核心需求，典型架构分为三层：

数据层：存储号码归属地、运营商、标签等结构化数据
服务层：提供查询接口、缓存管理、数据更新功能
接入层：处理HTTP请求、参数校验、结果封装

建议采用微服务架构，将号码查询、号码标记、数据更新拆分为独立服务。使用FastAPI框架可快速构建高性能API，其异步特性能有效处理并发请求。

# FastAPI基础示例
from fastapi import FastAPI
app = FastAPI()
@app.get("/query")
async def query_number(phone: str):
    # 实际实现需连接数据库
    return {"phone": phone, "region": "北京", "carrier": "移动"}

二、数据存储方案

1. 关系型数据库方案

MySQL/PostgreSQL适合存储结构化数据，表设计示例：

CREATE TABLE phone_data (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    phone_number VARCHAR(20) UNIQUE,
    province VARCHAR(20),
    city VARCHAR(20),
    carrier VARCHAR(20),
    spam_score INT DEFAULT 0,
    update_time TIMESTAMP
);

2. 内存数据库方案

Redis适合存储热点数据，可采用Hash结构：

import redis
r = redis.Redis(host='localhost', port=6379)
# 存储示例
r.hset("phone:13800138000", mapping={
    "province": "北京",
    "city": "北京",
    "carrier": "移动"
})

3. 搜索引擎方案

Elasticsearch适合模糊查询和全文检索，需建立索引映射：

PUT /phone_index
{
  "mappings": {
    "properties": {
      "phone_number": { "type": "keyword" },
      "province": { "type": "text" },
      "carrier": { "type": "keyword" }
    }
  }
}

三、核心功能实现

1. 号码解析模块

需处理多种号码格式，实现标准化的解析逻辑：

import re
def parse_phone(phone):
    # 去除空格、横线等分隔符
    cleaned = re.sub(r'[^\d]', '', phone)
    # 验证长度和前缀
    if len(cleaned) not in (11, 7, 8):
        raise ValueError("Invalid phone length")
    if not cleaned.startswith(('13', '14', '15', '16', '17', '18', '19')):
        raise ValueError("Invalid phone prefix")
    return cleaned[-7:]  # 返回后7位作为索引键

2. 查询接口实现

结合缓存和数据库的查询流程：

async def query_phone(phone: str):
    # 1. 先查Redis缓存
    cached = r.hgetall(f"phone:{phone}")
    if cached:
        return {"source": "cache", **cached}
    # 2. 查MySQL
    try:
        # 实际应使用异步ORM如SQLAlchemy
        result = db.query("SELECT * FROM phone_data WHERE phone_number=?", phone)
        if result:
            # 写入缓存，设置1小时过期
            r.hmset(f"phone:{phone}", mapping={
                "province": result["province"],
                "city": result["city"],
                "carrier": result["carrier"]
            })
            r.expire(f"phone:{phone}", 3600)
            return {"source": "db", **result}
    except Exception as e:
        log.error(f"DB query failed: {e}")
    # 3. 调用第三方API（可选）
    api_result = await call_thirdparty_api(phone)
    if api_result:
        return {"source": "api", **api_result}
    return {"error": "Not found"}

四、性能优化策略

1. 缓存策略设计

多级缓存：本地缓存（LRU）+ 分布式缓存（Redis）
缓存预热：系统启动时加载高频号码
缓存淘汰：基于LRU或LFU算法

2. 数据库优化

建立适当索引：ALTER TABLE phone_data ADD INDEX idx_phone (phone_number)
分库分表：按号码前缀进行水平拆分
读写分离：主库写，从库读

3. 异步处理

使用异步IO处理耗时操作：

import asyncio
from httpx import AsyncClient
async def call_thirdparty_api(phone):
    async with AsyncClient() as client:
        try:
            resp = await client.get(f"https://api.example.com/query?phone={phone}")
            return resp.json()
        except Exception:
            return None

五、扩展功能建议

号码标记系统：允许用户标记骚扰电话

@app.post("/mark")
async def mark_phone(phone: str, label: str):
    # 更新Redis中的标记计数
    current = r.hget(f"phone:{phone}", "spam_score") or 0
    new_score = min(int(current) + 1, 5)  # 限制最大分值
    r.hset(f"phone:{phone}", "spam_score", new_score)
    return {"status": "success"}

批量查询接口：支持一次查询多个号码
数据更新机制：定时从运营商更新号码库
安全防护：限制单位时间查询次数，防止爬虫

六、部署与监控

容器化部署：使用Docker打包服务

FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

监控指标：
- 查询成功率
- 平均响应时间
- 缓存命中率
- 数据库连接数
日志管理：结构化记录查询日志，便于分析

七、最佳实践总结

数据分层：热点数据放Redis，全量数据存MySQL
异步优先：非阻塞处理所有I/O操作
容错设计：每个依赖组件都应有降级方案
渐进更新：采用蓝绿部署或金丝雀发布更新服务
压力测试：使用Locust等工具模拟高并发场景

通过以上架构设计和实现策略，可构建出支持每秒数千QPS的高性能电话号码查询系统。实际开发中需根据业务规模调整各组件配置，例如小型系统可省略Elasticsearch，直接使用Redis+MySQL组合。

Python电话号码查询系统设计与实现指南