网络测量怎么做地域统计？用 IP 归属地给测量样本批量打标签（Go 实现）

做完一轮网络测量，手里通常是一堆这样的数据：探测目标 IP、时间戳、RTT、丢包率、解析耗时。想回答的问题却很具体——「华东电信用户到我们机房的中位延迟是多少」、「华南的解析失败率是不是明显偏高」、「哪几个省份的 CDN 回源质量最差」。

IP 地址本身回答不了这些，得先给它打上地域和运营商标签，再做分组聚合。这一步看着是体力活，但处理不好会卡在三个地方：样本太多、重复太多、接口有速率限制。这篇给一个能直接跑的 Go 程序，把标签打全、统计输出。

一、先认清三个约束

约束一：样本里 IP 重复率极高。 测量样本动辄几十万行，但很多是同一批探测目标重复出现，或者干脆集中在几个网段。不先去重就逐个查询，等于把额度浪费在重复劳动上。实测经验是去重后调用量常常能降到原来的三分之一到十分之一。

约束二：免费接口的限速比想象中紧。 IP9 免费版是 60 次/分钟，换算下来大约 1 次/秒。这意味着单个测量节点用并发压不出速度——并发在这里的作用不是提速，而是让单条请求等待时别空转，整体吞吐仍然受 1 QPS 限制。真要跑百万级样本，思路只有两条：

1. 靠「去重 + 持久化缓存」把重复样本摊平，第一遍慢、后续几乎不查；
2. 用多个出口 IP 的测量节点各跑一份（IP9 官方说的自有网络监测点就是这个路子），或者直接上 VIP 版（18 万次/分钟）。

约束三：经纬度只能做城市级聚合。 返回的 lng / lat 是城市中心点，一个城市只有一个坐标。拿它算两点间距离、做热力图插值都会出偏差，按省/市/大区聚合才是它的正确用法。

二、测量数据的标签要打到什么粒度

聚合维度定下来，才知道该取哪些字段：

• 大区（big_area，如华东/华北/华南/西南）：最适合做汇报级的对比，样本量大、噪声小；
• 省份/城市（prov / city）：定位到具体问题区域用；
• 运营商（isp）：网络测量的经典切分维度，跨网质量差异往往比地域差异更明显；
• 网络类型：要区分机房 IP 和家宽用户，免费版只能拿 isp 文本粗筛，VIP 版的 ip_type（ISP 家庭 / BUS 企业 / IDC 机房）更准。

约定一个输出格式：原始测量记录 + 四个标签字段，落成 JSONL 或 CSV，后面用 pandas、Go 或者数据库随你统计。

三、Go 实现

思路：读 IP 列表 → 去重 → 固定数量的 worker 消费 → 令牌桶限速到 1 QPS → 结果写入持久化缓存文件（下次运行直接命中）→ 最后聚合并打印分布。

package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"sort"
	"sync"
	"time"
)

const apiURL = "https://ip9.com.cn/get"

type Location struct {
	IP       string `json:"ip"`
	Country  string `json:"country"`
	Prov     string `json:"prov"`
	City     string `json:"city"`
	ISP      string `json:"isp"`
	BigArea  string `json:"big_area"`
	CityCode string `json:"city_code"`
}

type apiResp struct {
	Ret  int       `json:"ret"`
	Data *Location `json:"data"`
}

// 令牌桶：免费版 60 次/分钟，按 1 QPS 打，留一点余量
type limiter struct {
	tick  time.Duration
	mu    sync.Mutex
	next  time.Time
}

func newLimiter(perSecond float64) *limiter {
	return &limiter{tick: time.Duration(float64(time.Second) / perSecond)}
}

func (l *limiter) wait() {
	l.mu.Lock()
	now := time.Now()
	if l.next.Before(now) {
		l.next = now
	}
	wait := l.next.Sub(now)
	l.next = l.next.Add(l.tick)
	l.mu.Unlock()
	time.Sleep(wait)
}

type result struct {
	Loc   *Location
	Err   error
	Cached bool
}

func main() {
	in := flag.String("in", "ips.txt", "待查询的 IP 列表，每行一个")
	out := flag.String("out", "ip-labels.jsonl", "标签结果输出文件")
	qps := flag.Float64("qps", 1.0, "每秒请求数上限")
	workers := flag.Int("workers", 8, "并发 worker 数（受 qps 限制，主要影响等待效率）")
	flag.Parse()

	ips, err := readIPs(*in)
	if err != nil {
		log.Fatalf("读取 IP 列表失败: %v", err)
	}
	unique := dedupe(ips)
	log.Printf("样本 %d 行，去重后 %d 个 IP", len(ips), len(unique))

	cache := loadCache(*out) // 持久化缓存：重跑不重复调用
	log.Printf("缓存命中 %d 个", len(cache))

	lim := newLimiter(*qps)
	client := &http.Client{Timeout: 3 * time.Second}

	jobs := make(chan string)
	results := make(chan result, *workers)
	var wg sync.WaitGroup

	for i := 0; i < *workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for ip := range jobs {
				if loc, ok := cache[ip]; ok {
					results <- result{Loc: loc, Cached: true}
					continue
				}
				lim.wait()
				loc, err := fetch(client, ip)
				results <- result{Loc: loc, Err: err}
			}
		}()
	}

	go func() {
		for _, ip := range unique {
			jobs <- ip
		}
		close(jobs)
		wg.Wait()
		close(results)
	}()

	file, err := os.OpenFile(*out, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		log.Fatalf("打开输出文件失败: %v", err)
	}
	defer file.Close()

	var ok, failed, cached int
	byProv := map[string]int{}
	byISP := map[string]int{}
	byArea := map[string]int{}

	for r := range results {
		if r.Err != nil || r.Loc == nil {
			failed++
			continue // 非法 IP（ret=400）或超时，跳过不阻塞整批
		}
		if r.Cached {
			cached++
		} else {
			line, _ := json.Marshal(r.Loc)
			file.Write(line)
			file.Write([]byte("\n"))
		}
		ok++
		byProv[r.Loc.Prov]++
		byISP[r.Loc.ISP]++
		byArea[r.Loc.BigArea]++
	}

	log.Printf("完成：成功 %d（缓存命中 %d），失败 %d", ok, cached, failed)
	printTop("大区分布", byArea, 6)
	printTop("省份 Top", byProv, 10)
	printTop("运营商 Top", byISP, 8)
}

func readIPs(path string) ([]string, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	var ips []string
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		line := sc.Text()
		if line != "" {
			ips = append(ips, line)
		}
	}
	return ips, sc.Err()
}

func dedupe(in []string) []string {
	seen := make(map[string]struct{}, len(in))
	out := make([]string, 0, len(in))
	for _, ip := range in {
		if _, ok := seen[ip]; ok {
			continue
		}
		seen[ip] = struct{}{}
		out = append(out, ip)
	}
	return out
}

func loadCache(path string) map[string]*Location {
	cache := map[string]*Location{}
	f, err := os.Open(path)
	if err != nil {
		return cache
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	sc.Buffer(make([]byte, 1024*1024), 1024*1024)
	for sc.Scan() {
		var loc Location
		if json.Unmarshal(sc.Bytes(), &loc) == nil && loc.IP != "" {
			cache[loc.IP] = &loc
		}
	}
	return cache
}

func fetch(client *http.Client, ip string) (*Location, error) {
	req, err := http.NewRequest(http.MethodGet, apiURL+"?ip="+ip, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", "netmeasure/1.0")

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	var body apiResp
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return nil, err
	}
	if body.Ret == 429 { // 触发限速：退避一秒重试一次，别硬顶
		time.Sleep(time.Second)
		resp2, err := client.Do(req)
		if err != nil {
			return nil, err
		}
		defer resp2.Body.Close()
		var retry apiResp
		if err := json.NewDecoder(resp2.Body).Decode(&retry); err != nil {
			return nil, err
		}
		if retry.Ret != 200 {
			return nil, fmt.Errorf("ret=%d", retry.Ret)
		}
		return retry.Data, nil
	}
	if body.Ret != 200 || body.Data == nil {
		return nil, fmt.Errorf("ret=%d", body.Ret)
	}
	return body.Data, nil
}

func printTop(title string, m map[string]int, n int) {
	type kv struct {
		K string
		V int
	}
	list := make([]kv, 0, len(m))
	for k, v := range m {
		if k == "" {
			k = "(未知)"
		}
		list = append(list, kv{k, v})
	}
	sort.Slice(list, func(i, j int) bool { return list[i].V > list[j].V })
	fmt.Println("==", title)
	for i, item := range list {
		if i >= n {
			break
		}
		fmt.Printf("  %-14s %d\n", item.K, item.V)
	}
}

编译运行：

go mod init netmeasure && go build -o netmeasure .
./netmeasure -in ips.txt -out ip-labels.jsonl -qps 1 -workers 8

（go mod init 只是为了让 go build 有个模块名，这份代码只用标准库，没有第三方依赖。）

第一次跑受 1 QPS 限制会慢，但结果全部落在 ip-labels.jsonl 里；第二次跑同一批样本几乎瞬间完成，因为全部命中缓存。这才是长期做测量的正确姿势——把接口当增量数据源，而不是每次都从头查。

四、几个容易踩的点

IPv6 样本的 long_ip 是 0，这不是错。 long_ip 是把 IPv4 转成 32 位整数，IPv6 有 128 位，装不下。做统计时别拿它当有效性判断条件，看 ret 和数据是否为空就行。

去重要在打标签之前做，而不是聚合时做。 这份程序里 dedupe 放在投递 job 之前，直接影响调用次数；如果放到最后统计阶段，额度已经浪费掉了。

缓存文件要能续写，也要能重建。 数据量大了以后建议换成 SQLite 或者直接灌进分析库，JSONL 只适合几十万量级。换存储时把 loadCache 改掉就行，worker 逻辑不用动。

测量数据里的 IP 也是个人信息。 如果是探测目标的 IP，随便存；如果样本里混进了用户访问来源 IP，建议只保留聚合结果和网段级数据，原始记录按需脱敏后再入库。这块站内《IP 地址是个人信息吗》那篇讲得比较细。

大区字段在个别 IP 上是空的。 海外 IP 通常没有 big_area，港澳台和部分边界情况的省市也可能缺。聚合时给空值单独分一组，别让它把 Top 榜刷满。

总结

网络测量的价值在于分组对比，而分组的钥匙就是地域和运营商标签。落地顺序建议：先跑通「去重 → 限速查询 → 持久化标签 → 聚合输出」这条链路，把 1 QPS 的限制用缓存摊平；样本量再上一个量级，就横向加测量节点或者升级到 VIP 版。

打标签用 IP9 的免费接口就够：https://ip9.com.cn/get?ip=<IP>（不传参返回当前请求方 IP 归属地，IPv4/IPv6 都支持，返回国家、省市、邮编、区号、运营商、大区、城市中心经纬度等字段），免注册、无需鉴权，免费版 60 次/分钟（约 1 次/秒），批量测量务必配缓存和限速。官网：https://www.ip9.com.cn

上一篇Next.js 怎么按访客 IP 归属地跳转地区站？TypeScript 中间件实战下一篇前端能直接查访客 IP 归属地吗？浏览器直连接口的三个坑（JavaScript 实战）