PCDN 节点怎么识别？用 IP 归属地和流量特征排查私跑 PCDN（Go 实现）

某视频平台的带宽成本季度报表突然涨了 30%，机房出口流量曲线却看不出明显异常——再一查，大量内容请求没有命中 CDN 节点，而是被「分享」到了全国各地的家庭宽带用户家里。这些用户用闲置路由器和盒子跑着 PCDN 程序，把大流量从机房搬到了别人家。平台侧想把这些节点识别出来踢出调度，第一反应是「查 IP 归属地」，但拿到的字段全是省市和运营商，看不出任何问题——因为 PCDN 节点的 IP，本来就是最普通的家庭宽带 IP。

识别 PCDN 从来不是「查个归属地」就能完事，但 IP 归属地确实是整套判定里不可缺少的一环。这篇文章讲清楚平台侧怎么用「IP 归属地 + 流量特征」的组合把私跑 PCDN 的节点捞出来，给一段 Go 实现。

一、PCDN 节点为什么难认

PCDN（P2P CDN）把内容分发下沉到普通用户的家庭带宽：每个参与者的路由器、电视盒子、NAS 都变成「边缘节点」，上行带宽被用来给其他人供流。对内容平台来说这是省钱的路子，对运营商来说是破坏网络秩序的行为——国家整治 PCDN 的文件早就出了，各地运营商也在限速封堵。

难认在哪？三层：

1. IP 层无异常：节点 IP 归属地是正常城市、运营商是正常家宽运营商（电信/联通/移动），和普通用户一模一样；
2. 机房特征不存在：它不在 IDC，查 ASN 也是家庭宽带网络，用「机房 IP」规则直接漏光；
3. 只有流量维度露馅：长期高上行、连接数巨大、活跃时段固定（晚上睡觉也满负荷）——这些才是 PCDN 和马甲。

所以判定链是：先靠 IP 归属地/运营商给会话打上「家宽」标签，再叠加流量特征打分，两把尺子一起量。只量一把，要么全漏、要么全误伤。

二、方案设计

• 数据源：网关或交换机导出的连接日志（五元组 + 上下行字节数 + 同时连接数），不需要全部流量，抽样一小时足够；
• 特征分组：按「源 IP」聚合，计算上行/下行比、峰值并连接数、24 小时活跃度；
• 归属地打标：每个源 IP 查归属地和运营商，isp 字段区分家宽运营商（VIP 版还能拿 ip_type 直接分 ISP/BUS/IDC，识别更省事）；
• 打分规则：家宽 + 上行占比 > 60% + 并连接数 > 阈值 + 夜间活跃 → 嫌疑分累加，超过阈值进名单；
• 输出：嫌疑节点 Top 表（IP、归属地、运营商、特征指标），交给运维核对。

免费版 IP 接口 60 次/分钟/IP，批量查必须限速 + 缓存。下面的 Go 程序里配额按 50 次/分钟留余量，并发用一个 worker 池压住。

三、Go 实现

package main

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

// 连接日志一行: 源IP 上行字节 下行字节 并连接数 时段
// 例: 112.10.22.33 9830400 2457600 128 night

type NodeInfo struct {
	Up   int64 `json:"up"`
	Down int64 `json:"down"`
	Conn int   `json:"conn"`
}

type Geo struct {
	Prov string `json:"prov"`
	City string `json:"city"`
	ISP  string `json:"isp"`
}

const (
	apiBase   = "https://ip9.com.cn/get?ip="
	maxQPS    = 50 // 免费版 60 次/分钟，留 10 次余量
	suspectTh = 3  // 嫌疑分阈值
)

var (
	geoCache  = sync.Map{}
	rateLimit = time.Tick(time.Second / maxQPS)
)

func queryGeo(ip string) *Geo {
	if v, ok := geoCache.Load(ip); ok {
		return v.(*Geo)
	}
	<-rateLimit // 全局限速
	client := http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(apiBase + ip)
	if err != nil {
		return nil
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	var d struct {
		Ret  int `json:"ret"`
		Data Geo `json:"data"`
	}
	if err := json.Unmarshal(body, &d); err != nil || d.Ret != 200 {
		return nil // 非法 IP 或限速，跳过
	}
	geoCache.Store(ip, &d.Data)
	return &d.Data
}

// 判断家宽：运营商字段命中三大运营商且非专线/机房特征。
func isResidentialISP(isp string) bool {
	for _, kw := range []string{"电信", "联通", "移动", "广电"} {
		if strings.Contains(isp, kw) {
			return true
		}
	}
	return false
}

// 特征打分：每命中一条加 1 分。
func scoreNode(g *Geo, n *NodeInfo) int {
	s := 0
	if isResidentialISP(g.ISP) {
		s++
	}
	if n.Up > 0 && n.Up > n.Down*3 { // 上行远大于下行：纯供流特征
		s++
	}
	if n.Conn > 500 { // 并连接数远超普通家宽上网
		s++
	}
	return s
}

func main() {
	logPath := flag.String("log", "conn.log", "连接日志路径")
	flag.Parse()

	// 1. 按源 IP 聚合连接日志
	agg := map[string]*NodeInfo{}
	f, err := os.Open(*logPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "打开日志失败:", err)
		os.Exit(1)
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		parts := strings.Fields(sc.Text())
		if len(parts) < 5 {
			continue
		}
		var up, down int64
		var conn int
		fmt.Sscanf(parts[1], "%d", &up)
		fmt.Sscanf(parts[2], "%d", &down)
		fmt.Sscanf(parts[3], "%d", &conn)
		k := parts[0]
		if _, ok := agg[k]; !ok {
			agg[k] = &NodeInfo{}
		}
		agg[k].Up += up
		agg[k].Down += down
		if conn > agg[k].Conn {
			agg[k].Conn = conn
		}
	}

	// 2. 并发查 IP 归属地（worker 池）
	type item struct {
		ip string
		n  *NodeInfo
	}
	ips := make([]item, 0, len(agg))
	for ip, n := range agg {
		ips = append(ips, item{ip, n})
	}
	results := make(chan struct {
		ip string
		g  *Geo
		n  *NodeInfo
	}, len(ips))
	var wg sync.WaitGroup
	for i := 0; i < 8; i++ { // 8 个 worker，全局限速在 queryGeo 内
		wg.Add(1)
		go func() {
			defer wg.Done()
			for _, it := range ips {
				results <- struct {
					ip string
					g  *Geo
					n  *NodeInfo
				}{it.ip, queryGeo(it.ip), it.n}
			}
		}()
	}
	go func() { wg.Wait(); close(results) }()

	// 3. 打分排序
	type row struct {
		ip    string
		geo   *Geo
		up    int64
		down  int64
		conn  int
		score int
	}
	var rows []row
	for r := range results {
		if r.g == nil {
			continue
		}
		s := scoreNode(r.g, r.n)
		if s >= suspectTh {
			rows = append(rows, row{r.ip, r.g, r.n.Up, r.n.Down, r.n.Conn, s})
		}
	}
	sort.Slice(rows, func(i, j int) bool { return rows[i].score > rows[j].score })

	// 4. 输出嫌疑名单
	fmt.Printf("%-16s %-8s %-8s %-6s %10s %10s %6s %s\n",
		"IP", "省份", "城市", "运营商", "上行MB", "下行MB", "连接", "分")
	for _, r := range rows {
		fmt.Printf("%-16s %-8s %-8s %-6s %10.1f %10.1f %6d %d\n",
			r.ip, r.geo.Prov, r.geo.City, r.geo.ISP,
			float64(r.up)/1048576, float64(r.down)/1048576, r.conn, r.score)
	}
	fmt.Printf("\n嫌疑节点 %d 个\n", len(rows))
}

代码里三层逻辑对应开头说的判定链：isResidentialISP 用归属地接口的运营商字段把节点分成「家宽/非家宽」，scoreNode 叠加流量特征打分，最后的名单让运维人工复核。worker 池开的 8 个并发不会冲破限速，因为真正的节流在 queryGeo 里的全局 rateLimit。

四、落地注意事项

• 归属地是辅助，别当主判据：PCDN 判定必须带流量维度，只靠归属地等于没查；反过来只靠流量特征，公司 VPN 出口、办公室大内网上行也会误伤；
• 阈值先粗后细：上行比 3 倍、连接数 500 是保守初值，先跑一周样本看误报率再调，宁可先漏后严；
• 时段的权重：夜间（23 点—6 点）满负荷上行是最硬的 PCDN 信号，评分里可以单独加权重；
• 合规提示：识别到的是「疑似私跑 PCDN 的家庭宽带」，处理方式应是对接运营商或按用户协议处置，别公开点名；判断依据留日志，方便后续举证。

总结

PCDN 节点长得和普通家庭用户一模一样，识别靠两把尺子：归属地接口（免费版 https://ip9.com.cn/get?ip=xxx 的运营商/省市字段，VIP 版再加 ip_type 区分 ISP/BUS/IDC）打「家宽」标签，流量特征（上行比、并连接数、活跃时段）打「供流」标签，两把尺子一起量再打分。上面这段 Go 代码可以直接跑：喂一小时连接日志，出来一张嫌疑节点表。先试跑一周，把误报率调下来，再接进调度系统做节点摘除，比人工翻日志高效得多。官网：https://www.ip9.com.cn

上一篇问卷怎么按地区控制样本量？IP 归属地自动配额抽样（PHP 实现）下一篇为什么有些视频在别的地区看不了？IP 属地做版权分区的原理（Node.js 网关实现）