amqp（AMQP协议扩展）==源码级解析

amqp（AMQP协议扩展）

GitHub： https://github.com/php-amqp/php-amqp

完整标题

1. amqp源码全景：RabbitMQ的PHP C扩展——librabbitmq的封装
2. amqp扩展入口amqp.c：MINIT阶段的类注册与常量定义
3. AMQPConnection：连接管理的C实现——TCP连接与AMQP握手
4. AMQPChannel：通道的C实现——多路复用与通道隔离
5. AMQPQueue：队列操作的C实现——声明/绑定/消费/ack
6. AMQPExchange：交换器操作的C实现——direct/topic/fanout
7. AMQP消息发布：basic_publish的C实现——消息属性与路由键
8. AMQP消息消费：basic_consume的C实现——阻塞与非阻塞模式
9. AMQP事务：tx_select/tx_commit/tx_rollback的C实现
10. AMQP Publisher Confirm：发布确认的C实现——消息可靠投递

写到最深的标题

11. amqp的librabbitmq集成：amqp_rpc_reply_t的C处理——AMQP协议帧的解析与错误码映射
12. amqp的持久连接：跨请求的连接复用——AMQPConnection在persistent_list中的注册
13. amqp的SSL/TLS：amqp_ssl_socket_set_cacert的C调用——证书验证与加密通道

先立两个全篇约定,后面不再重复:
1.─这是"与 php-amqp 源码实现思路一致的代表性代码",不是逐字节照抄。字段名、方法签名随你─pin─的─librabbitmq─/─php–amqp──
版本浮动,机制是对的,精确拼写以你装的那个版本为准。
2. 编码/命名遵循 PHP 7/8 的 Zend API 与 librabbitmq 的惯例,函数名(如 amqp_basic_publish)是库的真实 API。

—–

总览:amqp 到底是个什么东西

大白话: php–amqp 是一个用 C 写的 PHP 扩展,它不自己实现 AMQP 协议,而是封装 RabbitMQ 官方用 C 写的客户端库
librabbitmq。librabbitmq 负责"把 AMQP 协议帧拼出来、发出去、收回来解析";php–amqp 负责"把这些 C 结构体包装成 PHP 的 OOP
对象(AMQPConnection、AMQPChannel、AMQPQueue、AMQPExchange、AMQPEnvelope)给你用"。

┌─────────────────────────────────────────────────┐
│ 你的 PHP 代码 │
│ $q = new AMQPQueue($channel); $q->consume(...); │
└─────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ php–amqp 扩展(C) │
│ 把 PHP 对象 <-> librabbitmq 的 C 结构体互转 │
│ amqp_connection_resource / amqp_channel_t / │
│ amqp_envelope_t / amqp_basic_properties_t │
└─────────────────────────────────────────────────┘
↓调用
┌─────────────────────────────────────────────────┐
│ librabbitmq(C) │
│ amqp_basic_publish / amqp_queue_declare / │
│ amqp_login ... 负责协议帧的编解码 │
└─────────────────────────────────────────────────┘
↓AMQP 08 协议(8 字节头 AMQP\\x00\\x00\\x09\\x01 + 帧)
┌─────────────────────────────────────────────────┐
│ RabbitMQ Server │
└─────────────────────────────────────────────────┘

为什么这样分层是核心议题:把"协议帧的编解码"这个又苦又容易错的部分丢给成熟的 C 库(librabbitmq),php–amqp 只管"PHP 对象 ↔
C 结构体"的转换层。这是它比纯 PHP 版 AMQP 快一个数量级的根本原因——没有一层PHP 代码在逐帧解释协议,全是原生 C 在做网络
I/O 和二进制解析。

—–

1. amqp 源码全景:librabbitmq 的封装层

大白话: 整个扩展就是一条"转换栈"。目录结构把"每个 PHP 类对应一个 C 文件"组织得清清楚楚。

php–amqp/
├── amqp.c ←模块入口:MINT/RSHUTDOWN/常量/类注册
├── amqp.h ←头文件、模块全局变量(AMQP_G)
├── php_amqp.h ←宏定义、版本号、全局变量声明
├── amqp_connection.c/.h ←AMQPConnection 类
├── amqp_channel.c/.h ←AMQPChannel 类
├── amqp_queue.c/.h ←AMQPQueue 类
├── amqp_exchange.c/.h ←AMQPExchange 类
├── amqp_envelope.c/.h ←AMQPEnvelope 类(消息信封)
├── amqp_basic_properties.c/.h←AMQPBasicProperties 类(消息属性)
├── amqp_exception.c/.h ←AMQPException 及子类
├── amqp_debug.c/.h ←调试日志钩子(amqp_log)
├── amqp_helpers.c/.h ←对象↔结构体互转、错误码映射
├── config.m4 ←构建脚本(探测 librabbitmq)
└── config.w32 ←Windows 构建

大白话再看一遍分层逻辑: 每个 .c 文件 = 一个 PHP 类;amqp_helpers.c = 通用转换工具;amqp.c =
模块的"大门口"。find"哪个类在哪个文件"就是找对应文件名。

—–

2. amqp.c:MINIT 阶段的类注册与常量定义

大白话: MINIT(Module INIT)是扩展被加载进 PHP(首次)时执行一次的函数。amqp.c 在这里干两件大事:①把所有 PHP
常量注册进去;②把每个类注册进 Zend 引擎。

完整代码(模块入口 + MINIT):

#include "php_amqp.h"

ZEND_BEGIN_MODULE_GLOBALS(amqp)
/* 模块级全局变量,比如持久连接的哈希表 */
HashTable persistent_connections;
ZEND_END_MODULE_GLOBALS(amqp)

/* 常量注册是 MINIT 的核心活 */
PHP_MINIT_FUNCTION(amqp)
{
zend_class_entry ce;

/* ── ①全局常量:flags 按位标志 ── */
REGISTER_LONG_CONSTANT("AMQP_NOPARAM", 0x00000000, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_DURABLE", 0x00000001, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_PASSIVE", 0x00000002, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_EXCLUSIVE", 0x00000004, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_AUTODELETE",0x00000008, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_MANDATORY", 0x00000040, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_IMMEDIATE", 0x00000080, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_IFUNUSED", 0x00000100, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_IFEMPTY", 0x00000200, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_MULTIPLE", 0x00000400, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_NOWAIT", 0x00000800, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_INTERNAL", 0x00001000, CONST_CS | CONST_PERSISTENT);
REGISTER_LONG_CONSTANT("AMQP_NOSIGNAL", 0x00002000, CONST_CS | CONST_PERSISTENT);

/* ── ②注册类:每个类来自各自的 .c 文件 ── */
INIT_CLASS_ENTRY(ce, "AMQPConnection", amqp_connection_methods);
amqp_connection_class_entry = zend_register_internal_class(&ce);
amqp_connection_register_class_constants(amqp_connection_class_entry);

amqp_register_exception_class("AMQPException", amqp_exception_class_entry, NULL);
amqp_register_exception_class("AMQPConnectionException",
amqp_connection_exception_class_entry, amqp_exception_class_entry);
amqp_register_exception_class("AMQPChannelException",
amqp_channel_exception_class_entry, amqp_exception_class_entry);
/* … QueueException / ExchangeException 同理 … */

return SUCCESS;
}

/* 模块入口表:告诉 Zend 哪个函数是 MINIT/MINFO 等 */
zend_module_entry amqp_module_entry = {
STANDARD_MODULE_HEADER,
"amqp",
NULL, /* functions:模块级函数(本扩展没有) */
PHP_MINIT(amqp),
PHP_MSHUTDOWN(amqp),
PHP_RINIT(amqp), /* 每次请求刚开始时执行 */
PHP_RSHUTDOWN(amqp),
PHP_MINFO(amqp),
PHP_AMQP_VERSION,
STANDARD_MODULE_PROPERTIES
};

大白话点拨:
– REGISTER_LONG_CONSTANT 用了 CONST_PERSISTENT,因为常量要在多次请求间保持,不能每次请求重注册。
– 那些 AMQP_* 常量就是你在 PHP 里写 $exchange->declare($exchange, 'direct', AMQP_DURABLE | AMQP_AUTODELETE)
传的位标志。它们本质就是整数位掩码,C 里用 | 组合,最后作为一个 uint32 flags 传给 librabbitmq。
– 用"每个类的 function entry 数组"(amqp_connection_methods)注册类,是 Zend 注册内建类的标准做法(不是用扩展机制)。

—–

3. AMQPConnection:连接管理的 C 实现

大白话: AMQPConnection 的职责是建立一条到 RabbitMQ 的 TCP 连接并完成 AMQP 握手(login)。它底层的状态是 librabbitmq 的
amqp_connection_state_t,php–amqp 用一个自定义结构包一层,再把结构体指针塞进 PHP 对象里。

完整代码:

/* php-amqp 在 C 层存连接状态的结构 */
typedef struct _amqp_connection_resource {
amqp_connection_state_t connection; /* librabbitmq 的连接句柄 */
amqp_connection_object *object; /* 反向引用 */
char *login; char *password; char *host; char *vhost;
int port;
int read_timeout; int write_timeout; int connect_timeout;
/* TLS 相关 */
int uses_tls;
char *cacert; char *cert; char *key;
zend_long heartbeat; /* 心跳间隔 */
unsigned int connected:1;
unsigned int persistent:1;
} amqp_connection_resource;

/* PHP 对象:就是包一个指向上面结构的指针 */
struct amqp_connection_object {
amqp_connection_resource *resource;
};

/* 核心:connect() 的 C 实现 */
static void amqp_connection_connect(amqp_connection_object *object)
{
amqp_connection_resource *res = object->resource;
amqp_connection_state_t conn = amqp_new_connection(); /* ①建句柄 */
res->connection = conn;

/* ②建立 socket(看是否走 TLS) */
if (res->uses_tls) {
amqp_ssl_socket_t *ssl = amqp_ssl_socket_new(conn);
amqp_ssl_socket_set_cacert(ssl, res->cacert); /* 见第 13 节 */
amqp_ssl_socket_set_key(ssl, res->cert, res->key);
amqp_ssl_socket_set_verify(ssl, AMQP_SSLVERIFY_PEER);
amqp_socket_open(ssl, res->host, res->port);
} else {
amqp_socket_t *sock = amqp_tcp_socket_new(conn);
amqp_socket_open(sock, res->host, res->port);
}

/* ③AMQP 握手 / login(就是发 login 帧,收 reply) */
amqp_rpc_reply_t reply = amqp_login(conn, res->vhost,
res->login, res->password,
AMQP_DEFAULT_CHANNEL_MAX, AMQP_DEFAULT_FRAME_MAX,
(uint16_t)res->heartbeat, AMQP_SASL_PLAIN);

/* ④检查握手结果,失败抛异常(见第 11 节) */
if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
amqp_exception_from_rpc_reply_throw(reply);
return;
}
res->connected = 1;
}

大白话点拨:
– amqp_new_connection() 只分配句柄,不发网络包;真正的网络连接发生在 amqp_socket_open。所以"TCP 连上"和"AMQP
握手成功"是两步,amqp_login 才代表"服务器接受了你"(用户名/密码/vhost 校验通过)。
– 心跳(heartbeat)在 login 时一并协商:AMQP_DEFAULT_FRAME_MAX(默认 131072=128KB)和 heartbeat
是协议层协商的值,双方取较小者。

为什么 TCP 连接不用自己写 socket? 因为 librabbitmq 已经把"帧编解码 + socket 读写下层"做掉了,php–amqp
只需要管"生命周期和对象包装"。这也是它又慢不了的省心点——所有网络I/O 都在 C 层,不经过 PHP 用户态。

—–

4. AMQPChannel:通道的 C 实现

大白话: 一个 AMQP 连接上可以开很多通道(channel),每条通道是独立的逻辑会话。AMQP 协议用一个 TCP
连接复用多路(multiplexing),靠
channel_id(1~65535)区分。通道隔离意味着:一条通道上的错误/事务/队列操作,不影响同一连接上的其他通道。AMQPChannel
就是包一个 channel_id + 反向引用所属连接。

完整代码:

struct amqp_channel_object {
amqp_channel_resource *resource; /* 内含 channel_id 和连接引用 */
};

typedef struct _amqp_channel_resource {
amqp_connection_object *connection; /* 反向引用:通道属于哪个连接 */
uint16_t channel_id; /* 通道号 */
int is_open:1;
} amqp_channel_resource;

/* 打开通道:发 channel.open 帧 */
static void amqp_channel_open(amqp_channel_resource *rsc)
{
amqp_connection_state_t conn = rsc->connection->resource->connection;
amqp_rpc_reply_t reply = amqp_channel_open(conn, rsc->channel_id);
if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
amqp_throw_amqp_channel_exception(reply);
return;
}
rsc->is_open = 1;
/* 通道建立后经常顺手设置 QoS */
amqp_basic_qos(conn, rsc->channel_id, 0, rsc->prefetch_count, 0);
}

/* 方法示例:prefetch(告诉 broker 一次最多推多少消息给你) */
PHP_METHOD(AMQPChannel, setPrefetchCount)
{
amqp_channel_object *obj = Z_AMQPCHANNEL_P(ZEND_THIS);
zend_long count;
ZEND_PARSE_PARAMETERS_START(1, 1)
Z_PARAM_LONG(count)
ZEND_PARSE_PARAMETERS_END();

obj->resource->prefetch_count = count;
amqp_basic_qos(obj->resource->connection->resource->connection,
obj->resource->channel_id, 0, count, 0);
}

大白话点拨:
– 多路复用的关键: 所有 AMQP 帧都带一个 channel 字段,broker 按 channel_id 把消息路由到正确的会话。所以"多路复用"不靠
PHP 层做线程,而是协议本身的设计——PH里只是"一个连接对象,多个 channel 对象,各自记 channel_id"。
– 通道隔离的含义: 第 9 节的事务(在通道上选事务模式)、第 10 节的确认模式,都是通道级状态。A 通道开了事务,B
通道照常发消息,互不影响。这就是"通道级隔离"最实用的体现。

—–

5. AMQPQueue:队列操作的 C 实现

大白话: AMQPQueue 把队列的生命周期操作(声明、绑定、消费、ack)包成方法。它底层就是调 librabbitmq 一排 amqp_queue_* /
amqp_basic_* 函数。

完整代码:

/* declare:声明队列,不存在则建,存在则返回信息 */
PHP_METHOD(AMQPQueue, declareQueue)
{
amqp_queue_object *obj = Z_AMQPQUEUE_P(ZEND_THIS);
amqp_channel_resource *ch = obj->resource->channel->resource;
amqp_connection_state_t conn = ch->connection->resource->connection;

amqp_queue_declare_ok_t *ok = amqp_queue_declare(conn,
ch->channel_id,
amqp_cstring_bytes(obj->resource->name), /* 队列名 */
obj->resource->flags & AMQP_PASSIVE, /* passive */
obj->resource->flags & AMQP_DURABLE, /* durable */
obj->resource->flags & AMQP_EXCLUSIVE, /* exclusive */
obj->resource->flags & AMQP_AUTODELETE, /* auto_delete */
0, /* nowait */
NULL); /* arguments */
if (ok) {
/* reply 里带了 broker 实际分配的队列名和消息数 */
obj->resource->name = amqp_cstring(ok->queue);
}
}

/* bind:绑定队列到交换器,通过路由键 */
PHP_METHOD(AMQPQueue, bind)
{
amqp_queue_bind(conn, ch->channel_id,
amqp_cstring_bytes(name), /* 队列 */
amqp_cstring_bytes(exchange), /* 交换器 */
amqp_cstring_bytes(routing_key), /* 路由键 */
nowait, NULL);
}

/* get:非阻塞拿一条(amqp_basic_get) */
PHP_METHOD(AMQPQueue, get)
{
amqp_rpc_reply_t reply = amqp_basic_get(conn, ch->channel_id,
amqp_cstring_bytes(name), no_ack);
/* 把返回的 amqp_envelope_t 包成 AMQPEnvelope 对象 */
if (reply.reply_type == AMQP_RESPONSE_NORMAL) {
amqp_envelope_t envelope;
amqp_message_t message;
if (amqp_consume_message(conn, &envelope, &message, NULL)) {
/* 转成 PHP 对象,见第 8 节 */
}
}
}

/* consume:阻塞式消费(带回调,见第 8 节) */
PHP_METHOD(AMQPQueue, consume)
{
amqp_basic_consume(conn, ch->channel_id, queue, consumer_tag,
no_local, no_ack, exclusive, 0, NULL);
/* 然后进入一个循环,一帧一帧地收消息并回调 PHP 用户函数 */
}

/* ack:确认消息 */
PHP_METHOD(AMQPQueue, ack)
{
amqp_basic_ack(conn, ch->channel_id, delivery_tag, multiple);
}

大白话点拨:
– ack 为什么常见坑: amqp_basic_ack 用 delivery_tag 确认"第几条"消息。如果忘了 ack,broker
会认为消息没消费成功,留在队列里,客户端重连后可能重复收到。所以"手动 ack"模式下,delivery_tag
要精确取自己要确认的那条(AMQPEnvelope::getDeliveryTag())。

—–

6. AMQPExchange:交换器操作的 C 实现

大白话: 交换器(exchange)是消息的路由中枢。AMQPExchange 负责交换器声明、绑定(到别的交换器)和最常用的
publish(发布消息)。类型 direct/topic/fanout/headers 本质就一个字符串("direct"/"topic"/fanout/headers),在声明时传给
broker。

完整代码:

/* 声明交换器 */
PHP_METHOD(AMQPExchange, declareExchange)
{
amqp_exchange_declare(conn, ch->channel_id,
amqp_cstring_bytes(name),
amqp_cstring_bytes(type), /* "direct" / "topic" / "fanout" / "headers" */
passive, durable, auto_delete, internal, nowait, NULL);
}

/* publish:发布消息 ——第 7 节展开 */
PHP_METHOD(AMQPExchange, publish)
{
amqp_basic_properties_t props;
amqp_basic_properties_init(&props);

/* 把 PHP 传进来的 body 转成 librabbitmq 的字节 */
amqp_bytes_t body = amqp_cstring_bytes(message);

/* 组装属性(见第 7 节) */
amqp_basic_publish(conn, ch->channel_id,
amqp_cstring_bytes(exchange),
amqp_cstring_bytes(routing_key),
flags, /* AMQP_MANDATORY | AMQP_IMMEDIATE */
&props, /* 属性 */
body);
}

大白话点拨:
– publish 里 exchange 为空串 "" 表示"发布到默认交换器"——此时routing_key
就是要投递到的队列名。这就是很多人"我明明只用了队列没建交换器"也能发消息的原因。
– direct:routing_key 精确匹配队列绑定键;topic:支持通配符(* 匹配一段,# 匹配多段);fanout:完全忽略
routing_key,发给所有绑定队列。这四个类型只是声明时的字符串,真正路由逻辑全在 broker 侧。

—–

7. AMQP 消息发布:basic_publish 的 C 实现

大白话: 发布一条消息 = 组装"属性(header)" + "正文(body)",然后一个 amqp_basic_publish 把帧写出去。消息属性就是
AMQPBasicProperties 对象的 C 结构 amqp_basic_properties_t。

完整代码:

PHP_METHOD(AMQPBasicProperties, __construct) { /* … */ }

static amqp_basic_properties_t amqp_basic_properties_from_object(
zval *zprops)
{
amqp_basic_properties_t props;
amqp_basic_properties_init(&props); /* 先全部清零 */

/* 关键:amqp_basic_properties_t 用"哪几个字段被设置了"的位掩码标记 */
uint32_t flags = 0;

/* 从 PHP 对象逐字段读出,能对应就填进 C 结构 */
if (content_type) {
props.content_type = amqp_cstring_bytes(content_type);
flags |= AMQP_BASIC_CONTENT_TYPE_FLAG;
}
if (delivery_mode) {
props.delivery_mode = delivery_mode; /* 1=transient / 2=persistent */
flags |= AMQP_BASIC_DELIVERY_MODE_FLAG;
}
if (headers) {
props.headers = amqp_table_t_from_zval(headers); /* 键值表 */
flags |= AMQP_BASIC_HEADERS_FLAG;
}
/* priority / correlation_id / reply_to / expiration / message_id /
timestamp / type / user_id / app_id … 同理 */

props._flags = flags; /* 告诉 librabbitmq 哪些字段有效 */
return props;
}

PHP_METHOD(AMQPExchange, publish)
{
amqp_basic_properties_t props = amqp_basic_properties_from_object(z_props);

amqp_basic_publish(conn, ch->channel_id,
amqp_cstring_bytes(exchange),
amqp_cstring_bytes(routing_key),
flags, &props,
amqp_cstring_bytes(body));
}

大白话点拨:
– _flags 是这个结构最关键的地方。 librabbitmq 的 amqp_basic_properties_t 是个大结构,但很多字段是可选的。它用 _flags
位掩码标"哪些字段真的被设置了",这样编解码时只序列化有效的字段。php–amqp 的核心工作之一,就是把 PHP
对象里的属性转成这个带 flags 的 C 结构。
– delivery_mode=2(persistent) 就是"消息落盘",broker
重启也不丢;=1(transient)只存内存。这是消息可靠性的第一层,但真正的可靠投递要配第 10 节的发布确认。

为什么消息属性要做成单独一个 AMQPBasicProperties 类,而不是直接塞进 publish 的参数里?
因为属性字段多达十几项(正文类型、优先级、关联 ID、过期时间、应用
ID、头表……),做成参数列表会又长又乱;做成对象可以按需设置、可复用、可序列化。这是为可读性牺牲了一点性能的设计。

—–

8. AMQP 消息消费:basic_consume 的 C 实现

大白话: 消费分两种:①非阻塞的get——发一个basic.get,broker 若有消息立刻返回,没有就返回"空";②阻塞的consume——发一个
basic.consume,然后停在 C 层循环里持续收帧,每收一条就回调你把 PHP 函数,直到通道/连接关闭或 cancel。

完整代码:

PHP_METHOD(AMQPQueue, consume)
{
amqp_basic_consume(conn, ch->channel_id,
amqp_cstring_bytes(queue_name),
amqp_cstring_bytes(consumer_tag),
no_local, no_ack, exclusive, 0, NULL);

/* 阻塞循环:一帧一帧收,直到回调返回 false / 超时 / 连接断开 */
while (1) {
amqp_envelope_t envelope;
amqp_message_t message;
amqp_rpc_reply_t r = amqp_consume_message(conn, &envelope, &message, &timeout);

if (r.reply_type == AMQP_RESPONSE_NORMAL) {
/* 把 C 的 envelope 包成 AMQPEnvelope 的 PHP 对象 */
zval zenv;
amqp_envelope_to_zval(&envelope, &zenv);

/* 调你传进来的 PHP 回调 */
zval retval;
zend_call_function_with_1_param(callback, &zenv, &retval);
zval_ptr_dtor(&zenv);

/* 回调返回 false 就停 */
if (Z_TYPE(retval) == IS_FALSE) {
zval_ptr_dtor(&retval);
break;
}
zval_ptr_dtor(&retval);
} else if (r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) {
/* broker 把连接/通道关了 */
break;
}
amqp_basic_destroy_envelope(&envelope);
amqp_destroy_message(&message);
}

amqp_basic_cancel(conn, ch->channel_id, amqp_cstring_bytes(consumer_tag));
}

/* 非阻塞 get */
PHP_METHOD(AMQPQueue, get)
{
if (amqp_basic_get(conn, ch->channel_id, queue, no_ack)) {
/* 有消息就返回 AMQPEnvelope */
amqp_envelope_t envelope;
amqp_consume_message(conn, &envelope, &message, NULL);
return envelope_to_zval(&envelope);
}
RETURN_FALSE; /* 没消息 */
}

大白话点拨:
– consume 阻塞意味着你的 PHP 进程"卡"在 C 循环里。 这是它和 get 的本质区别:consume 会一直挂着等消息,不释放 CPU;get
是一问一答,没消息就返回 false。所以生产环境用 consume + 回调;需要批处理或短任务用 get。
– 回调返回 false 是消费循环的"退出开关"——php–amq约定用户回调返回 false 就停止消费。这是库给用户的一种控制手段。
– amqp_consume_message 是 librabbitmq 的阻塞收单条消息函数,它内部处理帧、heartbeat、channel 关闭等。php–amqp
只是在外面套循环。

—–

9. AMQP 事务:tx_select / tx_commit / tx_rollback 的 C 实现

大白话: 事务是一个通道级的选项:在通道上 tx_select 开启事务模式,之后该通道上的 publish/ack 会被"缓冲",要么 tx_commit
一次性提交,要么 tx_rollback 全部回滚。注意事务只能覆盖单条通道,跨通道做不到。

完整代码:

PHP_METHOD(AMQPChannel, startTransaction)
{
amqp_tx_select(conn, ch->channel_id); /* 发 tx.select */
/* tx_select 也是 RPC,要等 reply */
amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);
amqp_check_rpc_reply_throw(reply);
}

PHP_METHOD(AMQPChannel, commitTransaction)
{
amqp_tx_commit(conn, ch->channel_id); /* 发 tx.commit */
amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);
amqp_check_rpc_reply_throw(reply);
}

PHP_METHOD(AMQPChannel, rollbackTransaction)
{
amqp_tx_rollback(conn, ch->channel_id); /* 发 tx.rollback */
amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);
amqp_check_rpc_reply_throw(reply);
}

大白话点拨:
– 每次 startTransaction 后,这一条通道上的所有发布都会暂存在 broker 侧(broker 内存),直到 commitTransaction
才真正排队,或 rollbackTransaction 丢弃。所以事务能保证"这一批要么全到、要么全不到"。
– 事务是"通道级",不是"连接级"、更不是"客户端级"。这意味着:你要事务,就得在单独一条通道上做,别和其他消息混同一通道。
– 这是最强的 trade–off: 事务模式下消息先缓冲,broker 不能立即路由,吞吐明显下降。RabbitMQ 官方在能做到 publisher
confirm(下一节)之后,其实不推荐用事务做常规发布,因为确认机制能提供类似的可靠性,却不用牺牲缓冲带来的吞吐。

—–

10. AMQP Publisher Confirm:发布确认的 C 实现

大白话: 事务是"all-or-nothing"但要缓冲;发布确认是"每个消息发出去,broker 回我一个 OK"。通道进入 confirm 模式后,broker
对每条 publish 都会回一个 basic.ack(确认)或 basic.nack(拒绝)。这样你就确切知道哪条消息 broker 收下了。

完整代码:

/* ①把通道切到 confirm 模式 */
PHP_METHOD(AMQPChannel, confirmSelect)
{
amqp_confirm_select(conn, ch->channel_id); /* 发 confirm.select */
amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);
amqp_check_rpc_reply_throw(reply);
ch->resource->confirm_mode = 1;
}

/* ②发布若干条消息后,等确认 / 检查结果 */
PHP_METHOD(AMQPChannel, waitForConfirm)
{
amqp_connection_state_t conn = ch->connection->resource->connection;

while (ch->resource->unconfirmed_count > 0) {
/* 阻塞收帧,直到 broker 返回 basic.ack / basic.nack */
amqp_frame_t frame;
amqp_rpc_reply_t r = amqp_get_frame(conn, &frame);

if (frame.frame_type == AMQP_FRAME_METHOD) {
if (frame.payload.method.id == AMQP_BASIC_ACK_METHOD) {
ch->resource->unconfirmed_count—;
/* 通知用户:这条 delivery_tag 已确认 */
call_ack_callback(ch, frame.payload.method.decoded.basic_ack.delivery_tag);
} else if (frame.payload.method.id == AMQP_BASIC_NACK_METHOD) {
/* nack = broker 明确说这条没接住 */
call_nack_callback(ch, ...);
}
}
}
}

大白话点拨:
– 和事务的区别一句话说清: 事务在"发送前"存起来、最后统一交;确认是在"发送时"就发出去、等 broker
逐条回话。所以确认不牺牲吞吐(不用缓冲整批),但要求你等回话——等不到就等于可能丢了。
– 可靠投递的完整链条是:①delivery_mode=2(落盘)+②confirm模式(等 broker 说收到)+ ③消费端手动
ack(处理完再确认)。三步一起才是"恰好一次/可靠一次"的工程实践;单独一步都不够。
– 实际工程里,confirm 的坑在于如何处理未确认的中间态:通常做法是"发了 N 条 →waitForConfirm →若有 nack
或超时未确认,就把没确认的重新入队或补偿"。这需要你在业务层做幂等处理,因为 confirm
不能帮你保证"恰好一次"(只能保证"broker 确实收到"这条)。

—–

11. librabbitmq 集成:amqp_rpc_reply_t 的 C 处理

大白话: 这是把"库的抽象"翻译成"PHP 能看到的东西"的关键一层。几乎每个 AMQP 操作都是"发一个请求帧 →等一个响应帧"的
RPC。librabbitmq 把所有 RPC 的响应都统一包成一个 amqp_rpc_reply_t,php–amqp 根据它的类型判断成败、取出错误码、映射成
PHP 异常。

完整代码:

/* librabbitmq 的 RPC 回复结构(真实定义,简化) */
typedef struct amqp_rpc_reply_t_ {
amqp_response_type reply_type; /* NORMAL / LIBRARY_EXCEPTION / SERVER_EXCEPTION */
union {
amqp_library_error_t lib; /* 客户端库自身的错误(连不上/超时/内存) */
amqp_connection_close_t connection_close; /* broker 关连接 */
amqp_channel_close_t channel_close; /* broker 关通道 */
/* 这两个 close 结构里都有 reply_code(uint16) 和 reply_text(amqp_bytes_t) */
} reply;
} amqp_rpc_reply_t;

/* php-amqp 的统一检查函数 */
void amqp_check_rpc_reply_throw(amqp_rpc_reply_t *reply)
{
if (reply->reply_type == AMQP_RESPONSE_NORMAL) {
return; /* 一切正常 */
}

uint32_t code;
const char *msg;

if (reply->reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) {
/* 库级错误:比如 TCP 断开、握手超时、协议错乱 */
code = reply->reply.lib.code;
msg = amqp_error_string2(code); /* librabbitmq 自带的错误文本 */
zend_throw_exception(amqp_connection_exception_class_entry,
msg, code);
} else if (reply->reply_type == AMQP_RESPONSE_SERVER_EXCEPTION) {
/* broker 级错误:这里有真正的 reply_code,要逐码翻译成人类可读 */
code = reply->reply.connection_close.reply_code; /* 或 channel_close */
msg = reply->reply.connection_close.reply_text.len
? (char*)reply->reply.connection_close.reply_text.bytes
: "Unknown server error";

/* 逐码映射,增加可读性 */
switch (code) {
case AMQP_CONNECTION_FORCED: msg = "Connection was closed by broker"; break;
case AMQP_NOT_FOUND: msg = "Not found (queue/exchange/vhost)"; break;
case AMQP_ACCESS_REFUSED: msg = "Access refused (credentials/vhost)"; break;
case AMQP_CHANNEL_ERROR: msg = "Channel error"; break;
/* … many more … */
}
zend_throw_exception(amqp_exception_class_entry, msg, code);
}
}

大白话点拨:
– reply_type 就是"成败的第一判据",只分三类: 正常(NORMAL)、库自己错了(LIBRARY)、服务器拒绝了(SERVER)。其中 SDK 复杂度的
90% 都藏在后两类。
– 错误码映射是它的增值点: librabbitmq 只给你一个原始 code 和一句话,php–amqp 把这些标准 AMQP reply–code(404 not
found、403 access refused、320 connection forced、406 precondition
failed……)翻译成更直观的英文,并抛成对应异常类(连接类错误抛AMQPConnectionException,通道类抛
AMQPChannelException)。这正是"封装"值的钱——库的原始报错对使用PHP 的人不友好。

—–

12. 持久连接:跨请求的连接复用

大白话: 普通 connect() 是"每个请求建一条新连接、请求结束就断",这对短任务/Web
请求来说,握手(connect+login)的成本占比很高。pconnect()(persistent connect)让连接跨多个 PHP
请求复用——第一次建好后,塞进模块全局的哈希表;后续请求按参数找表、命中就直接复用,省掉重复握手。

完整代码:

/* 在 amqp.h 里,模块全局里有个哈希表存持久连接 */
ZEND_BEGIN_MODULE_GLOBALS(amqp)
HashTable persistent_connections;
ZEND_END_MODULE_GLOBALS(amqp)

/* 生成一个"连接指纹",键 = 主机+端口+vhost+用户(足以区分会话) */
static char *amqp_connection_key(amqp_connection_resource *res)
{
return strpprintf(0, "%s:%d:%s:%s", res->host, res->port,
res->vhost, res->login);
}

PHP_METHOD(AMQPConnection, pconnect)
{
char *key = amqp_connection_key(res);

/* ①先查表:命中就直接复用,不再握手 */
zval *existing = zend_hash_str_find(&AMQP_G(persistent_connections), key, strlen(key));
if (existing) {
amqp_connection_resource *old = Z_AMQPCONNECTION_P(existing)->resource;
if (old->connected) {
/* 复用:把当前对象指向已存在的连接,引用计数 +1 */
object->resource = old;
return;
}
/* 否则视为失效,删掉重建 */
zend_hash_str_del(&AMQP_G(persistent_connections), key, strlen(key));
}

/* ②没命中,新建一条并放进表 */
amqp_connection_connect(object);
zend_hash_str_update(&AMQP_G(persistent_connections), key, strlen(key), obj);

/* ③请求结束时(RSHUTDOWN)不真正关闭,而是减引用计数 */
}

PHP_RSHUTDOWN_FUNCTION(amqp)
{
/* 遍历 persistent_connections:引用计数 >0 就不关,留给下个请求;
若请求没正常释放,才真正 disconnect */
return SUCCESS;
}

大白话点拨:
– 持久连接的本质 = 用一个"连接指纹"做 key 的全局缓存。 指纹由连接参数拼接而成,保证"同一个
vhost/账号/主机"的连接才能复用,不会张冠李戴。
– 引用计数是生命线: 因为连接跨请求存在,必须知道"还有几个请求在用",才能决定"最后一个用完再关"。这是持久连接最容易写错
bug 的地方——漏减引用计数会导致连接泄漏(每个请求多出一个没断的连接),最终把RabbitMQ 的连接数撑爆。
– 持久连接和 PHP–FPM 的 worker 模型天然配对——进程常驻,连接也跟着常驻。但对短脚本/ CLIm
下的一次性任务,持久连接没意义(进程结束连接必断)。

—–

13. SSL/TLS:amqp_ssl_socket_set_cacert 的 C 调用

大白话: 走 TLS 时,php–amqp 不是手动开一个加密 socket,而是用 librabbitmq 的 amqp_ssl_socket_* 系列函数,它内部基于
OpenSSL 建一条加密的 TCP 通道。握手(AMQP 8
字节协议头)是在这条加密通道之上进行的。证书校验(CA、主机名)都在这几步里配置。

完整代码:

/* 在 connect() 的 TLS 分支里,完整展开(对应第 3 节) */
amqp_ssl_socket_t *ssl = amqp_ssl_socket_new(conn); /* ①建 SSL socket 句柄 */

/* ②CA 证书:用来验证服务端证书是不是你信任的 CA 签的 */
if (res->cacert) {
amqp_ssl_socket_set_cacert(ssl, res->cacert);
}

/* ③可选:客户端证书 + 私钥(双向 TLS / mTLS,服务端要认证你) */
if (res->cert && res->key) {
amqp_ssl_socket_set_key(ssl, res->cert, res->key);
}

/* ④验证强度:是否校验对端证书 / 是否校验主机名 */
amqp_ssl_socket_set_verify(ssl, AMQP_SSLVERIFY_PEER); /* 校验证书链 */
amqp_ssl_socket_set_verify_name(ssl, AMQP_SSLVERIFY_PEER_NAME); /* 校验主机名 */

/* ⑤真正建连接(解密层在此之上) */
amqp_socket_open(ssl, res->host, res->port);

/* 之后 amqp_login 走的就是加密通道 */

大白话点拨:
– AMQP_SSLVERIFY_* 分两档: 只校验证书链(PEER)vs 连主机名一起校验(PEER_NAME)。生产环境一定开
PEER_NAME,否则"证书合法但不属于这个域名"也能连,中间人攻击仍有机会。AMQP_SSLVERIFY_NONE 只建议在本地调试用。
– set_cacert 指 CA 文件,set_key 指客户端证书+私钥(仅 mTLS)。大部分场景只要 cacert(确认服务端身份)+
verify_name(校验域名)。
– 为什么加密放在 librabbitmq 而不是 php–amqp? 因为 SSL 握手、证书验证、加解密是"传输层"的事,librabbitmq 自己封装了
OpenSSL,php–amqp 只需把 cacert/cert/key/verify 这几个选项透传进去。这再次体现分层:php–amqp
只做"配置的搬运工",重活交给库。

—–

三个设问(统一回答)

1. 为什么选"封装 librabbitmq"这个方案,而不是别的?

一句话: 因为 AMQP 协议的"帧编解码 + 网络 I/O"是这个领域最容易做错又最不值得自己重做的部分,而 RabbitMQ 官方已经用 C
把它做成了成熟库(librabbitmq),php–amqp 选择了站在它的肩膀上。

对比的可行替代,各自输在哪:

┌────────────────────────────┬────────────────────────────────────────────────────────────────────────────────────┐
│ 方案 │ 结局 │
├────────────────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
│ 纯 PHP 实现 AMQP │ 每一条消息都要 PHP 层逐字节解析协议,CPU 和内存开销高一个量级;长时间消费时 PHP 的 │
│ 协议(如旧版 php–amqplib) │ GC 和内存模型会很吃力。纯 PHP 版的定位是"能用",而 C 版定位是"快"。 │
├────────────────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
│ 直接自己写 socket + │ 自己实现 = 自己背 AMQP 各种状态机(连接、通道、handshake │
│ 协议帧(不依赖 librabbitmq) │ 超时、heartbeat、异常帧)的锅。重复造轮子,还大概率造得比官方差。 │
├────────────────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
│ 用别的语言写绑定(如 Rust │ 可行,但要额外引入 FFI / 外部可执行,分发(见上一批 PIE │
│ 的 amqprs 做个 PHP FFI 桥) │ 系列)、编译链复杂度飙升。对"要快又要好发"的 PHP 扩展,直接用 C 封装 C 库最顺。 │
└────────────────────────────┴────────────────────────────────────────────────────────────────────────────────────┘

php–amqp 真正的设计决定是"把协议层外包,自己只做 OOP 包装层"。 代价是它强依赖 librabbitmq 的 API 和 ABI,librabbitmq
升级、API 变动,php–amqp 就得跟着适配(历史上 RabbitMQ 3.x 的握手/心跳演进就动过 librabbitmq 的底层)。

2. 这个方案的 trade–off 是什么?

– 强耦合 librabbitmq / 平台工具链。 编译要
librabbitmq–dev(或源码),部署要匹配系统架构;跨平台(macOS/Windows/Linux)各有一套构建,分发麻烦(这正是上一批 PIE
想解决的痛点)。
– 原生化带来的"调试地狱"。 出问题就是 C 层 segfault / ASan 报告(见上一批 B2),不做 ASan 时极其难排查;amqp_rpc_reply_t
一层一层的结构体,报错不如纯 PHP 库友好。
– Zend ABI 版本碎片。 每个 PHP 大版本(8.1/8.2/8.3)ABI 不同,得分别编译;php–amqp 自身也要跟着 Zend
内部结构(如对象模型、zend_object 升级)适配。
– 阻塞模型对并发不友好。 consume() 是阻塞式 C 循环,一个 PHP worker 挂在一条通道上等消息期间,干不了别的活。FPM 下要靠多
worker 幂等;协作并发要么用事件循环(不在 C 扩展里),要么只在 CLI 常驻进程里用。
– 封装层也有"翻译成本"。 "C 结构体 ↔PHP 对象"的每一次互转(尤其 amqp_basic_properties_t 那种十几字段、带 _flags
的结构)都会产生额外开销。对超大消息/超高频 publish,这个转换本身就是瓶颈之一(不过仍远低于纯 PHP 解析)。
– 版权/维护归属。 librabbitmq 是 MIT,php–amqp 是 PHP–3.01,但库的升级节奏、ABI 稳定性不由 php–amqp
说了算——依赖方被上游绑架。

3. 如果让我重新设计

内核我不改:继续封装 librabbitmq,继续走"协议帧交给 C 库、PHP 只做 OOP
包装"这条路——这是性能和正确性都最稳的选择。我会改四件事,每一件都冲着上面的trade–off 去:

1. 把"阻塞"还给用户,提供事件驱动模式。 现版consume()是"一条通道、一个 worker、卡死在 C 循环"。我会加一个
非阻塞的"拉取一帧/一组"API + 可插拔事件循环接口,让同一个连接能在 N 条通道上轮询取消息,而不是 N 个 worker
各自阻塞。这样 FPM/长任务都能高效,不用靠多进程堆并发。
2. 把"C 结构体 ↔PHP 对象"的转换缓存/复用。 amqp_basic_properties_t 和 amqp_envelope_t
的互转目前是每次新建、每次析构。我会在模块全局做对象池,尤其对高频
publish,让属性结构体复用,只改字段,把"翻译成本"压到最低。
3. 给错误码引入真正的结构化异常层级,而不是一个带 code 的 message。 现在错误都抛成 AMQPException(code, msg),code
是一个裸整数。我会定义:连接断/超时 →可重试类异常;not found/access refused →不重试类异常;让调用方一 catch
就能区分"要不要重试",而不是自己 switch code。这对"保证可靠投递"的工程实践(第 10 节)帮助最大。
4. 把持久连接做成显式池,并加入健康检查。 现在 pconnect 靠一个简单哈希 +
引用计数,断线后能否正确重建、会不会泄漏,是比较脆弱的点。我会改成带心跳探测 +
断线自愈的"连接池"——请求取用时如果发现连接已死,自动重连而不是罢工;引用计数+ 超时闲置回收一起做,避免把 RabbitMQ
连接数撑爆。

核心判断: php–amqp 最大的长处(底层是快而稳的 C 库)和最大的短板(阻塞模型 + 转换开销 +
弱错误语义)是一体两面。要优化的不是"换掉 librabbitmq",而是"让 C 层的能力对 PHP
使用者更友好"——把事件驱动、对象复用、结构化异常、连接池这四件事做好。