2.5 连接与通道：AMQP 的传输底座

2.5 连接与通道：AMQP 的传输底座

本节摘要：连接（Connection）是客户端与 Broker 之间的 TCP 长连接，通道（Channel）是连接内部的多路复用虚拟连接：所有 AMQP 动作都发生在通道上，通道随连接断开而失效。本节讲清这对底座的分工、心跳机制与故障形态，并复盘一次连接泄漏拖垮 Broker 的真实事故。

前面四节都在看消息的路由与存放，这一节往下一层，看承托它们的传输底座。这里有两个高频事故的故乡：连接泄漏与通道误用。底座不稳，上层的可靠性设计全是空中楼阁。

为什么需要两层

TCP 连接的建立与销毁是重操作：三次握手、TLS 协商（若启用）、AMQP 协议头交换，全程几十毫秒起步，且每个连接在 Broker 侧占用文件句柄与内存。而 AMQP 的日常操作——发布、确认、声明——每秒可能成千上万次。用 TCP 直接承载这些操作，要么性能被握手开销吃光，要么 Broker 被海量连接压垮。

AMQP 的答案是通道复用：一条 TCP 连接内部可以开出多条逻辑通道，每条通道有独立编号，帧上带着通道号在连接内交织传输。线程模型随之简化——多线程共享一条连接，各用各的通道，互不阻塞（主流客户端库对 BlockingConnection 这类同步接口限制一通道一线程，跨线程共用通道会收到串扰错误）。

两层结构画出来一目了然：

图 5 连接与通道的多路复用结构

图的右下角藏着一条排障黄金法则：故障域看层级。通道层的错误（参数冲突、权限不足）只影响一条通道，重建即愈；连接层的错误（网络断、心跳超时、认证失败）带走全部通道，必须重连。排查时先判断错误落在哪一层，处置方案完全不同——把通道级故障当连接级处理，会导致无谓的全量重连风暴，反而加重 Broker 负担。

两者的容量账要分开算。连接数受 Broker 端文件句柄限制，默认上限可以容纳数万，但单个连接的内存与心跳维护成本决定真实容量远低于此；通道数单连接可开数千，但每条通道占一份状态内存，且 Broker 的某些内部簿记按通道计数。实践口径：单进程一条连接加按需通道；线程数大的服务可以拆几条连接分摊，而不是几百条。

心跳：半开连接的探雷器

TCP 不主动通知对端死亡——网线拔了、机器断电了，连接状态在 Broker 眼里依然是"活着"。心跳机制解决这个谎报军情的问题：客户端与 Broker 约定间隔，定期互发心跳帧，连续两次收不到即判定对端失联，强制关闭连接并清理资源。

心跳间隔可以协商，pika 里是 ConnectionParameters 的 heartbeat 参数（秒）。设太短，网络抖动或 GC 停顿会误杀健康连接；设太长，死连接占着句柄不肯走。三十到六十秒是常见的稳妥区间，容器化环境里还要保证心跳间隔小于各层负载均衡的空闲超时，否则代理层会先掐掉"安静"的连接。

完整演练：一次连接泄漏事故的解剖

背景：某报表服务每次生成报表都新建连接发消息，用完不关。上线半年无事，直到一次大促：报表频率提升，连接数以每小时几百的速度爬升。当晚 Broker 内存告警，随后拒绝新连接——所有依赖它的服务跟着遭殃，一个"不发消息只收报表请求"的服务放倒了整个消息平台。

操作：事故复盘用的三段诊断。第一步，看连接数量与来源：

rabbitmqctl list_connections user peer_host state channels
# 预期输出（节选，连接数已过万）：
# report_svc	10.2.3.17	running	1
# report_svc	10.2.3.17	running	1
# report_svc	10.2.3.17	running	1
# order_svc	10.2.3.21	running	8
​

第二步，定位元凶特征：report_svc 的连接单连接单通道、数量持续增长，order_svc 少连接多通道——后者才是健康的复用形态。第三步，修复代码，从"每任务一连接"改为"进程内单例连接"：

import pika, threading

class Publisher:
    """进程级单例：一条连接，按线程取通道"""
    def __init__(self):
        self._conn = pika.BlockingConnection(
            pika.ConnectionParameters(host="mq.internal", heartbeat=30))
        self._local = threading.local()

    def channel(self):
        # 每线程一条通道，线程退出由调用方负责 close
        if not hasattr(self._local, "ch"):
            self._local.ch = self._conn.channel()
        return self._local.ch

    def publish(self, exchange, routing_key, body):
        self.channel().basic_publish(
            exchange=exchange, routing_key=routing_key, body=body)

publisher = Publisher()   # 模块级初始化，全进程共享
​

结果：连接数从两万多回落到常驻几十条，Broker 内存告警解除。解读：连接泄漏的隐蔽性在于它按业务量缓慢累积，监控上表现为一条缓慢爬升的斜线而不是突刺——这正是 4.4 节监控要盯"连接数趋势"而非只看当前值的原因。另外注意 ChannelClosedByBroker 这类通道级错误：通道死了连接还活着，正确处理是关闭旧通道、重开新通道，而不是推倒整个连接重建。

变式一：心跳实验。把 heartbeat 设为 1 秒，再在消费回调里 sleep 三秒模拟慢处理，观察连接被 Broker 判死——心跳与业务超时的关系比想象中微妙，pika 的 BlockingConnection 会在回调期间无法发送心跳，这正是长任务要拆分或用异步客户端的原因。变式二：验证通道隔离。两个线程共用一条通道做发布，复现帧串扰异常，再改回各用通道恢复正常。

客户端选型对底座的影响

pika 的 BlockingConnection 是同步阻塞模型，代码直观，但消费回调里的慢逻辑会拖死心跳，多线程支持也弱。Java 生态的 Spring AMQP、Python 的 aio-pika 等封装层自带你想要的池化与恢复逻辑：连接自动重连、通道重建、消费者重订阅。第 5 章的 Spring Boot 集成会展示这套机制的 Java 版本。选客户端时，"断线恢复是否自动"应当与"API 是否顺手"同权重。

⚠️ 常见坑：在消费回调里做长耗时处理。同步客户端回调期间心跳发不出去，处理超过两倍心跳间隔，连接就被 Broker 判死，随后消费确认发不出去，消息重新入队，形成"处理了等于白处理"的循环。长任务要么拆批，要么把心跳间隔调大并接受故障发现变慢，要么换异步客户端。

本节要点回顾

• 两层设计：连接承载 TCP 成本，通道承载操作频率，多路复用是性能根基；
• 容量口径：进程内单连接、按线程开通道，连接数以几十计而非几百；
• 心跳机制：定期互发心跳帧判死半开连接，间隔三十到六十秒为宜；
• 泄漏特征：单连接单通道且数量缓涨，监控盯趋势而非瞬时值；
• 通道级恢复：通道死亡不推倒连接，关闭重建通道即可。

底座夯实了。下一节把本章全部知识装回一次投递，做一次完整的路由全流程追踪——这是全册追踪单的中枢页。