Gnutella：一个被 AOL 杀死的项目，如何让数百万人免费下载 MP3

预计阅读时间：16 分钟

2000 年初，AOL 内部有人写了一个小东西——一个完全不需要中心服务器的文件共享协议演示。公司高层觉得这东西太危险，下令砍掉。但代码已经流出，互联网不认"撤回"按钮。几天之内，Gnutella 就像野草一样蔓延开来，几年后数百万人在用它下载 MP3。

这个故事本身就够戏剧性了，但更有意思的是：Gnutella 的协议设计极其简陋，甚至可以说是"粗暴"。它凭什么撑了十年？

五条消息，撑起一个网络

Gnutella 协议的核心只有五种消息类型：

消息	方向	作用
Ping	发起者 → 邻居 → 邻居的邻居…	"我在这里，谁活着？"
Pong	收到 Ping 的节点 → 沿原路返回	"我活着，我持有 X 个文件，Y MB"
Query	搜索者 → 广播转发	"谁有叫 'metallica' 的文件？"
QueryHit	拥有文件的节点 → 沿原路返回	"我有，IP 是 A，端口是 B"
Push	下载者 → 拥有文件的节点	"我防火墙挡了入站，你主动连我推送文件"

没有握手协商，没有认证，没有加密。每条消息前面加一个固定格式的头部（descriptor header），然后就是原始数据。节点收到消息后，根据头部里的 TTL 和 hops 决定是否继续转发——TTL 每经过一个节点减 1，减到 0 就丢弃。

这就是整个协议的路由机制。没有路由表，没有 DHT，没有结构化拓扑。消息就是靠广播一跳一跳往外扩散，像水波纹一样。

连接建立：一句问候定生死

两个 Gnutella 节点想成为邻居，过程非常原始：

发起方发送：
GNUTELLA CONNECT/0.6\r\n

接收方回复：
GNUTELLA OK/0.6\r\n

就这么一行字符串，通过 TCP 通道传输。如果对方回了 OK，连接建立，之后所有消息都用 Gnutella 自定义的二进制帧格式在这个 TCP 连接上双向流动。没有 HTTP，没有 JSON，没有版本协商——虽然 0.6 版本后来加了一些能力协商的扩展头部，但核心逻辑还是：你说 CONNECT，我说 OK，咱们就是邻居了。

每个节点通常维护 3-5 个这样的邻居连接。消息从任何一个邻居进来，广播到其他所有邻居出去——唯独不往来的方向回传，避免环路。

广播的代价：查询洪泛

Gnutella 最致命的设计缺陷就是查询方式。你想找一个文件，发一条 Query，这条消息会被广播到你能到达的所有节点。假设平均每个节点有 4 个邻居，TTL 设为 7，那一条 Query 在最坏情况下会被复制 4⁷ ≈ 16,384 次。

2000 年的拨号上网用户，带宽 56 kbps。当网络膨胀到几十万节点时，Query 洪泛直接把大量节点的带宽吃光。这就是著名的"free rider"问题的放大版——大量用户只下载不共享，少数共享节点被 Query 流量压垮。

后来社区做了几个补救措施：

动态查询（Dynamic Querying）：不再一次性广播高 TTL 的 Query，而是先发 TTL=1 的试探，如果没有足够结果，逐步提高 TTL。这把洪泛的量砍掉了一个数量级。
超节点（Ultrapeer）：带宽好、共享文件多的节点升为超节点，承担路由职责；弱节点只连超节点，不参与转发。网络从扁平结构变成两层，广播范围大幅缩小。
Query Routing Protocol (QRP)：每个叶子节点把自己拥有的文件名做成一个 Bloom filter 压缩后发给超节点。超节点收到 Query 时，先查 Bloom filter，只有可能命中才转发给该叶子节点。

这些补丁让 Gnutella 又多活了几年，但协议骨架里的广播基因始终没被根除。

用 Python 模拟一个最小 Gnutella 节点

下面这段代码实现了一个极简的 Gnutella 协议核心逻辑——握手、Ping/Pong、Query/QueryHit 的编码解码和广播转发。它不能连接真实的 Gnutella 网络（现代网络已经基本消失），但能让你亲手看到这些二进制消息是怎么构造和传播的。

#!/usr/bin/env python3
"""minimal_gnutella.py — 最小 Gnutella 协议演示节点

只实现了协议核心的消息编码/解码和模拟广播转发，
不连接真实网络。运行后会在本地模拟一个小型 P2P 网络，
展示 Ping/Pong 和 Query/QueryHit 的传播过程。

依赖：仅标准库
用法：python3 minimal_gnutella.py
"""

import struct
import hashlib
import random
import time
from collections import defaultdict

# ── Gnutella Descriptor Header 格式 ──
# | msg_id (16B) | type (1B) | ttl (1B) | hops (1B) | payload_len (4B) |
# 总共 23 字节，大端序

MSG_PING    = 0x00
MSG_PONG    = 0x01
MSG_QUERY   = 0x80
MSG_QUERYHIT = 0x81
MSG_PUSH    = 0x40

HEADER_SIZE = 23

def make_msg_id():
    """生成 16 字节随机消息 ID"""
    return hashlib.sha256(struct.pack('>d', time.time()) + bytes(random.getrandbits(8) for _ in range(8))).digest()[:16]

def pack_header(msg_id, msg_type, ttl, hops, payload_len):
    return struct.pack('>16sBBBBI', msg_id, msg_type, ttl, hops, payload_len)

def unpack_header(data):
    if len(data) < HEADER_SIZE:
        raise ValueError(f"Header too short: {len(data)} bytes")
    msg_id, msg_type, ttl, hops, payload_len = struct.unpack('>16sBBBBI', data[:HEADER_SIZE])
    return msg_id, msg_type, ttl, hops, payload_len

# ── Ping: payload 为空 ──
def make_ping(ttl=7):
    msg_id = make_msg_id()
    return pack_header(msg_id, MSG_PING, ttl, 0, 0), msg_id

# ── Pong: payload = ip(4B) + port(2B) + num_files(4B) + num_kb(4B) ──
def make_pong(msg_id, ip, port, num_files, num_kb, ttl=1, hops=0):
    payload = struct.pack('>IHIH', ip, port, num_files, num_kb)
    return pack_header(msg_id, MSG_PONG, ttl, hops, len(payload)) + payload

# ── Query: payload = min_speed(2B) + query_string(null-terminated) ──
def make_query(query_str, min_speed=0, ttl=7):
    msg_id = make_msg_id()
    payload = struct.pack('>H', min_speed) + query_str.encode('utf-8') + b'\x00'
    return pack_header(msg_id, MSG_QUERY, ttl, 0, len(payload)) + payload, msg_id

# ── QueryHit: payload = num_hits(1B) + port(2B) + ip(4B) + speed(4B)
#              + result_set (8B per hit: file_index(4B)+file_size(4B) + file_name\0)
#              + servant_id (16B)
# ──
def make_queryhit(query_msg_id, ip, port, speed, results, servant_id, ttl=1, hops=0):
    num_hits = len(results)
    payload = struct.pack('>BIHI', num_hits, port, ip, speed)
    for file_idx, file_size, file_name in results:
        payload += struct.pack('>II', file_idx, file_size) + file_name.encode('utf-8') + b'\x00'
    payload += servant_id
    return pack_header(query_msg_id, MSG_QUERYHIT, ttl, hops, len(payload)) + payload


# ── 模拟节点 ──
class SimNode:
    def __init__(self, name, ip, port, files):
        self.name = name
        self.ip = ip
        self.port = port
        self.files = files  # dict: {filename: (index, size_kb)}
        self.servant_id = make_msg_id()
        self.neighbors = []
        self.seen_msgs = set()  # 防止重复转发

    def connect(self, other):
        self.neighbors.append(other)
        other.neighbors.append(self)

    def receive(self, raw_msg, from_node=None):
        header = raw_msg[:HEADER_SIZE]
        msg_id, msg_type, ttl, hops, payload_len = unpack_header(header)
        payload = raw_msg[HEADER_SIZE:]

        # 已见过这条消息 → 丢弃（防环路）
        if msg_id in self.seen_msgs:
            return []
        self.seen_msgs.add(msg_id)

        outgoing = []

        if msg_type == MSG_PING:
            # 回复 Pong，沿原路返回（这里简化为直接回复给来源）
            pong = make_pong(msg_id, self.ip, self.port,
                             len(self.files), sum(s for _, s in self.files.values()))
            outgoing.append((pong, from_node))  # 回传给来源节点

            # 如果 TTL > 0，继续广播给其他邻居
            if ttl > 1:
                new_ttl = ttl - 1
                new_hops = hops + 1
                fwd_header = pack_header(msg_id, MSG_PING, new_ttl, new_hops, 0)
                for nb in self.neighbors:
                    if nb is not from_node:
                        outgoing.append((fwd_header, nb))

        elif msg_type == MSG_QUERY:
            min_speed = struct.unpack('>H', payload[:2])[0]
            query_str = payload[2:].rstrip(b'\x00').decode('utf-8')

            # 本地搜索
            hits = []
            for fname, (fidx, fsize) in self.files.items():
                if query_str.lower() in fname.lower():
                    hits.append((fidx, fsize, fname))

            if hits:
                qh = make_queryhit(msg_id, self.ip, self.port, 0, hits, self.servant_id)
                outgoing.append((qh, from_node))  # 沿原路返回

            # TTL > 0 → 转发给其他邻居
            if ttl > 1:
                new_ttl = ttl - 1
                new_hops = hops + 1
                fwd = pack_header(msg_id, MSG_QUERY, new_ttl, new_hops, payload_len) + payload
                for nb in self.neighbors:
                    if nb is not from_node:
                        outgoing.append((fwd, nb))

        elif msg_type == MSG_PONG or msg_type == MSG_QUERYHIT:
            # Pong 和 QueryHit 沿原路返回，不广播
            # 在模拟中我们直接传回给 from_node
            outgoing.append((raw_msg, from_node))

        return outgoing


# ── 模拟网络运行 ──
def simulate():
    # 创建 5 个节点，各持有不同文件
    nodes = [
        SimNode("Alice",   0x7F000001, 6346, {"metallica_one.mp3": (0, 4200), "radiohead_ok.mp3": (1, 5800)}),
        SimNode("Bob",     0x7F000002, 6347, {"beatles_letitbe.mp3": (0, 3100)}),
        SimNode("Carol",   0x7F000003, 6348, {"nirvana_smells.mp3": (0, 3600), "metallica_enter.mp3": (1, 4900)}),
        SimNode("Dave",    0x7F000004, 6349, {"pink_floyd_wall.mp3": (0, 7200)}),
        SimNode("Eve",     0x7F000005, 6350, {"radiohead_creep.mp3": (0, 3200), "nirvana_lithium.mp3": (1, 2800)}),
    ]

    # 建立邻居关系：Alice-Bob-Carol-Dave-Eve（链式拓扑）
    for i in range(len(nodes) - 1):
        nodes[i].connect(nodes[i + 1])

    print("=== Gnutella 最小协议模拟 ===")
    print(f"网络拓扑: {', '.join(n.name for n in nodes)} (链式)")
    print()

    # ① Alice 发 Ping，观察传播
    print("--- [1] Alice 发出 Ping (TTL=3) ---")
    ping_raw, ping_id = make_ping(ttl=3)
    deliver_queue = [(ping_raw, nodes[0], None)]  # (msg, receiver, sender)

    step = 0
    while deliver_queue and step < 50:
        step += 1
        msg, receiver, sender = deliver_queue.pop(0)
        sender_name = sender.name if sender else "外部"
        print(f"  → {receiver.name} 收到消息 (来自 {sender_name})")
        responses = receiver.receive(msg, from_node=sender)
        for resp_msg, target in responses:
            deliver_queue.append((resp_msg, target, receiver))

    print()

    # ② Alice 发 Query 搜索 "metallica"
    print("--- [2] Alice 发出 Query: 'metallica' (TTL=3) ---")
    # 重置 seen_msgs
    for n in nodes:
        n.seen_msgs = set()

    query_raw, query_id = make_query("metallica", ttl=3)
    deliver_queue = [(query_raw, nodes[0], None)]

    step = 0
    found_files = []
    while deliver_queue and step < 50:
        step += 1
        msg, receiver, sender = deliver_queue.pop(0)
        sender_name = sender.name if sender else "外部"
        msg_id, msg_type, ttl, hops, plen = unpack_header(msg[:HEADER_SIZE])

        type_name = {MSG_PING: "Ping", MSG_PONG: "Pong",
                     MSG_QUERY: "Query", MSG_QUERYHIT: "QueryHit"}.get(msg_type, f"0x{msg_type:02x}")

        print(f"  → {receiver.name} 收到 {type_name} (TTL={ttl}, hops={hops}, 来自 {sender_name})")

        if msg_type == MSG_QUERYHIT:
            payload = msg[HEADER_SIZE:]
            num_hits = payload[0]
            print(f"    ★ {receiver.name} 收到 QueryHit: {num_hits} 个匹配文件!")

        responses = receiver.receive(msg, from_node=sender)
        for resp_msg, target in responses:
            rtype = unpack_header(resp_msg[:HEADER_SIZE])[1]
            if rtype == MSG_QUERYHIT:
                found_files.append((target, resp_msg))
            deliver_queue.append((resp_msg, target, receiver))

    print()
    print("--- [3] 搜索结果汇总 ---")
    if found_files:
        for target_node, qh_msg in found_files:
            payload = qh_msg[HEADER_SIZE:]
            num_hits = payload[0]
            offset = 11  # 跳过 num_hits(1) + port(2) + ip(4) + speed(4)
            for _ in range(num_hits):
                fidx, fsize = struct.unpack('>II', payload[offset:offset+8])
                fname = payload[offset+8:].split(b'\x00')[0].decode('utf-8')
                print(f"  {target_node.name} 有文件: {fname} ({fsize} KB)")
                offset += 8 + len(fname.encode('utf-8')) + 1
    else:
        print("  未找到匹配文件")


if __name__ == '__main__':
    simulate()

运行方式：

python3 minimal_gnutella.py

你会看到 Ping 消息从 Alice 一跳一跳传到 Eve，Pong 沿原路回传；Query 搜索 "metallica" 广播出去后，Carol 的 QueryHit 沿路返回到 Alice。整个过程就是 2000 年那个协议的缩影——粗暴、简陋，但确实能跑。

"覆水难收"的深层原因

AOL 管层下令删除 Gnutella 代码时，他们犯了一个对互联网本质的误判：去中心化系统没有"关机按钮。

中心化服务（比如 Napster）有一个索引服务器——关掉服务器，网络就死了。法院命令 Napster 关机，Napster 就真的死了。但 Gnutella 没有任何中心组件。每个节点既是客户端也是服务器，任何人随时可以启动一个新节点、连上邻居、整个网络就又多了一块。你没法向法院申请"关掉 Gnutella"，因为不存在一个可以关的实体。

这也是为什么后来 BitTorrent 选择了类似路线——Tracker 可以被关掉，但 DHT 和 PEX 让无 Tracker 的种子照样能下载。去中心化不是技术选择，是生存策略。

十年长尾：为什么它最终衰退了

Gnutella 的衰退不是被某个竞争对手"杀死"的，而是自身设计天花板加上外部环境变化：

带宽效率：即使加了 Ultrapeer 和 QRP，广播查询的带宽开销仍然远高于 BitTorrent 的 DHT 结构化查询。当文件从几 MB 的 MP3 变成几百 MB 的电影，这个差距变得致命。
法律压力：虽然网络本身无法被关掉，但开发客户端软件的公司（比如 LimeWire）可以被起诉。2010 年 LimeWire 被法院强制关闭，大量用户流失。
用户体验：搜索结果混乱、下载速度不稳定、文件质量不可信——这些问题在 BitTorrent 的种子站点和评论系统中得到了更好的解决。

Gnutella 今天仍然有极少数节点在运行，属于互联网的"长尾遗迹"。它更像一个技术化石——提醒后来者：去中心化设计能赋予系统惊人的韧性，但简陋的协议架构终究会被更精细的设计取代。

如果你想自己实验

上面的模拟代码可以随意改造。几个值得尝试的方向：

把链式拓扑改成更真实的随机图，观察广播消息的扩散路径差异
把 TTL 从 3 改成 7，看消息量膨胀多少倍
加入 Ultrapeer 逻辑：让某些节点只转发不搜索，叶子节点只连超节点
用 Bloom filter 实现 QRP：叶子节点把文件名哈希进 filter，超节点用它过滤 Query

这些改动能让你亲手感受到：为什么原始 Gnutella 会在规模膨胀时崩溃，以及后来的补丁设计到底救了什么、没救什么。