前几个月因为业务需求我需要一个输入IP就能返回地理位置、运营商信息，甚至风险标签的工具。刚开始我觉得很简单，网上那么多免费网站，直接抓取一下不就行了？后来才发现，这条路走不通——免费网站要么限流，要么数据不准，更别提批量查询的时候根本达不到我需要的字段质量。后来我决定自己搭一套。这篇文章就是整个过程的复盘，希望对有类似需求的同行有所帮助。

第一步：选数据源，离线库还是在线API？

离线库与在线API各具优势，具体怎么选择可以参考一下这张对比表：

如果只是做个内部工具，日均查询量几百次，在线API就够了；如果要接入高并发业务，或者部署在隔离内网，离线数据库是最佳选择。这只是我的建议和看法，具体怎么选择还是看你的业务需求和想法。

我的需求就是要支撑一个日均万次查询的广告反刷系统，对比下来离线数据库更符合我的需求。付费IP离线库基本都包含地理位置、运营商、风险评分、代理标签等所需字段，更新频率也可以满足我们的需求。如此来看就是要看价格和精准度了，找了几家服务商进行报价和测试，最终敲定了IP数据云的IP归属地和风险画像库，数据精准符合需求，报价也完全符合我的心理预期，可以说是性价比杠杠的了。

第二步：架构设计，一个查询接口的骨架

敲定了数据源，下一步就可以开始设计服务架构了。为了方便理解这是一个简化后的架构图：

核心设计要点：

• 无状态服务：查询服务可以水平扩展，每台机器独立加载一份数据库到内存
• 缓存层：热点IP查询结果存入Redis，进一步降低数据库读取压力
• 热更新：每日凌晨从服务端拉取最新数据库文件，替换旧文件，服务无需重启

第三步：编码实现，一个完整的查询接口

我选择用Python + Flask实现，数据库使用MMDB格式，这是行业标准，查询效率极高。

import maxminddb
import redis
from flask import Flask, request, jsonify
import threading
import time
app = Flask(__name__)
# 加载IP数据库
db_path = "/data/ipdb/ipdb.mmdb"
reader = maxminddb.open_database(db_path)
# Redis连接
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def query_ip_location(ip):
    """查询IP归属地，带缓存"""
    # 先查缓存
    cached = redis_client.get(f"ip:{ip}")
    if cached:
        return eval(cached)  # 实际生产环境建议用json.loads
    
    # 查本地数据库
    data = reader.get(ip)
    if not data:
        return {"status": "not_found"}
    
    # 结构化返回数据
    result = {
        "ip": ip,
        "country": data.get("country"),
        "province": data.get("province"),
        "city": data.get("city"),
        "district": data.get("district"),
        "isp": data.get("isp"),
        "usage_type": data.get("usage_type"),
        "risk_score": data.get("risk_score"),
        "latitude": data.get("latitude"),
        "longitude": data.get("longitude")
    }
    
    # 写入缓存，过期时间1小时
    redis_client.setex(f"ip:{ip}", 3600, str(result))
    return result
@app.route('/ip/query')
def ip_query():
    ip = request.args.get('ip')
    if not ip:
        return jsonify({"error": "missing ip parameter"}), 400
    
    result = query_ip_location(ip)
    return jsonify(result)
@app.route('/ip/batch', methods=['POST'])
def ip_batch():
    """批量查询接口"""
    ips = request.json.get('ips', [])
    if not ips or len(ips) > 100:
        return jsonify({"error": "batch size must be between 1 and 100"}), 400
    
    results = {}
    for ip in ips:
        results[ip] = query_ip_location(ip)
    return jsonify(results)
def update_database():
    """每日更新数据库文件的线程"""
    while True:
        time.sleep(86400)  # 24小时
        # 这里调用下载脚本，下载最新的MMDB文件
        # 下载完成后，用新reader替换旧reader
        global reader
        new_reader = maxminddb.open_database("/data/ipdb/ipdb_new.mmdb")
        # 原子替换
        reader = new_reader
        print("数据库已更新")
if __name__ == '__main__':
    # 启动更新线程
    updater = threading.Thread(target=update_database, daemon=True)
    updater.start()
    app.run(host='0.0.0.0', port=8080)

• 单IP查询：通过 /ip/query?ip=xxx调用，返回结构化JSON
• 批量查询：通过 /ip/batch POST请求，支持一次最多100个IP
• Redis缓存：热点IP缓存1小时，大幅降低数据库压力
• 热更新：后台线程每日自动拉取最新数据库文件，无缝切换

第四步：性能测试与优化

部署上线前，我用 wrk做了压测。单台4核8G服务器，本地MMDB查询压测结果：

缓存效果非常明显。如果业务需要更高吞吐，可以增加实例数量，通过Nginx负载均衡。

第五步：部署运维的几个坑

1. 内存占用：MMDB文件本身200-400MB，Python加载后还会增加一些开销。建议留足1GB内存，避免OOM。

2. 数据更新：每日更新时，不要直接覆盖正在使用的文件。我的做法是下载到临时文件，验证完整性后再原子替换。

3. 缓存穿透：如果大量查询不存在的IP（比如内网地址），会绕过缓存直击数据库。可以在代码里加一个“不存在”的缓存标记，比如 {"status":"not_found"}也缓存几分钟。

4. IPv6支持：选数据库时务必确认是否支持IPv6。现在IPv6流量占比已经超过40%，不支持IPv6的数据库基本没法用。

第六步：IP查询工具的应用

系统一上线就被小伙伴接进了反刷系统，用来观察和判断广告曝光和点击的IP是否来自数据中心或代理，过滤无效服务器流量，有效抵御恶意点击和流量作弊，保障广告投放的真实有效。。

• 成果展示：识别出30%以上的虚假点击，挽回的广告预算损失完全可以覆盖当初数据库购买的费用。（四舍五入就相当于没花钱啊！思考.jpg）

总结

搭建一个IP归属地查询网站，技术上并不复杂：选好数据源、设计好架构、写对代码，就能稳定跑起来，而且Github也有不少开源软件可以获取，只是简单的IP归属地查询已经完全够用了。如果你也需要一个IP地址查询网站，不妨也动手试试吧。