README

workbunny/webman-rabbitmq

🐇 A pure-PHP implementation of AMQP for webman/workerman 🐇

📖 说明

此文档为 3.0 版本 2.0 LTS 版本请查看 2.0 文档

✨ 简介

适配 Workerman/webman 的 AMQP 组件包

核心特性

✅ 支持基于 AMQP 协议工具实现 AMQP-Server
✅ 支持 5 种消费模式：简单队列、workQueue、routing、pub/sub、exchange
✅ 支持延迟队列（RabbitMQ 须安装插件）
✅ 支持连接池、通道池，Builder 支持影子模式（并发补偿）
✅ 提供 QueueSimpleBuilder：消费与生产使用不同连接（TCP 隔离），避免消费回调阻塞导致生产侧心跳失效
✅ 3.0 版本相比之前版本，更符合 AMQP 协议约定，更合理的架构设计和使用逻辑
- 使用 ConnectionManagement 多连接管理器管理 Connection（Client），合理复用机制及并发使用能力
- 使用 Channel-Pool 管理 Channel，合理的复用和并发机制
- 提供 AMQP 协议包，可供开发者自定义实现 AMQP-Client 或 AMQP-Server，并提供 AMQP-Frame 协议帧工具

🏗️ 架构概念

    ┌───────────┐
    | Builder A | ──┐
    └───────────┘   |                                          | ┌───────────┐
                    |                                          | | Channel 1 |
                    |                                          | └───────────┘
    ┌───────────┐   └─> ┌──────────────────┐                   | ┌───────────┐
    | Builder A | ────> | Connections Pool | ── connection ──> | | Channel 2 |
    └───────────┘   ┌─> └──────────────────┘   min ... MAX     | └───────────┘
                    |         <static>          <context>      | ┌───────────┐
                    |                                          | | Channel 3 |
    ┌───────────┐   |                                          | └───────────┘
    | Builder C | ──┘                                                 ...
    └───────────┘                                                 channel-max

核心组件说明

🔨 Builder

队列消费者、生产者的抽象结构，类似 ORM 的 Model

BuilderConfig: 队列配置结构
Builder 可以指定不同的 connection 配置进行连接，以区分业务/服务
Builder 的 publish/consume 使用了影子模式（当前 Connection 的 Channel 耗尽时，会自动从 Connection Pool 获取新的连接创建 Channel）
- 影子模式下请尽量将 Connection Pool 和 Channels Pool 的配置 wait_timeout 改小，避免过长时间的等待（等待中会出让控制权，不会阻塞）

Builder 模式：

模式	类	消费/生产连接	说明
`queue`（默认）	`QueueBuilder`	共用	常规队列消费/生产，同一连接内以 `Channel` 区分
`simple`	`QueueSimpleBuilder`	隔离	消费独占一条连接且不归还；生产走连接池中另一条连接（详见消费与生产 TCP 隔离）

🔗 Connection

基于 AsyncTcpConnection 封装的 AMQP-client

Connection 由 ConnectionManagement 管理，连接池为静态，不会因为 Builder 的释放而释放
Connection Pool 中通过 get 拿取 Connection 后需要手动调用 release 归还，或者使用 action 通过传入回调函数来执行并自动归还
支持 pool-less 模式：设置 connections_pool.enable = false 后使用专用长连接，消除池借还竞态
配置信息：
- enable: 是否启用连接池
- min_connections: 最小连接数
- max_connections: 最大连接数
- idle_timeout: 空闲回收时间 [s]
- wait_timeout: 等待连接超时时间 [s]

📡 Channel

抽象的通道对象

每一个 Connection 都具备一个 Channel 池
channel() 支持两种模式：
- Context 模式（默认）：$connection->channel() — 同一协程复用，defer 归还（消费者专用）
- 回调模式：$connection->channel(false, fn($ch) => ...) — 借→用→立即归还（HTTP publish 等短生命周期场景）
配置信息：
- max_connections: 最大通道数 (null时使用server约定)
- idle_timeout: 空闲回收时间 [s]
- wait_timeout: 等待超时时间 [s]

📦 AMQP

workerman 支持的协议封装

📚 查看详细文档

📚 AMQP 0-9-1 协议帧对照表

🚀 快速开始

系统要求

PHP >= 8.1
webman-framework >= 2.0 或 workerman >= 5.1
rabbitmq-server >= 3.10

安装

composer require workbunny/webman-rabbitmq

⚙️ 配置

基础配置 `app.php`

<?php declare(strict_types=1);

return [
    'enable' => true,
    // 日志 LoggerInterface | LoggerInterface::class
    'logger' => null,
];

连接配置 `connections.php`

<?php declare(strict_types=1);

use Workbunny\WebmanRabbitMQ\Connection\Connection;

return [
    'default' => [
        'connection'       => Connection::class,
        // 连接池，用于支撑影子模式
        'connections_pool' => [
            // 是否启用连接池
            'enable'                => false,
            'min_connections'       => 1,
            'max_connections'       => 20,
            'idle_timeout'          => 60,
            'wait_timeout'          => 10,
        ],
        'config' => [
            'debug'              => false,
            'host'               => '127.0.0.1',
            'vhost'              => '/',
            'port'               => 5672,
            'username'           => 'guest',
            'password'           => 'guest',
            'mechanism'          => 'AMQPLAIN',
            'timeout'            => 10,
            // 重启间隔
            'restart_interval'   => 5,
            // 通道池
            'channels_pool'      => [
                // 最大通道数，默认根据server约定
                'max_connections'  => null,
                'idle_timeout'     => 60,
                'wait_timeout'     => 10,
            ],
            'client_properties' => [
                'name'     => 'workbunny/webman-rabbitmq',
                'version'  => InstalledVersions::getVersion('workbunny/webman-rabbitmq'),
            ],
            // 心跳回调 callable
            'heartbeat_callback' => function () {
            },

            // see https://www.workerman.net/doc/workerman/async-tcp-connection/construct.html
//            'context' => [
//                'ssl' => [
//                    'verify_peer'      => false,
//                    'verify_peer_name' => false,
//                ],
//            ]
        ],
    ],
];

🛠️ 命令行工具

# 构建 Builder
php webman workbunny:rabbitmq-builder -h

# 移除/关闭 Builder
php webman workbunny:rabbitmq-remove -h

# 查看 Builder 列表
php webman workbunny:rabbitmq-list -h

⏰ 延迟队列

⚠️ 延迟队列需要为 RabbitMQ 安装 rabbitmq_delayed_message_exchange 插件

安装插件

进入 RabbitMQ 的 plugins 目录下执行命令下载插件（以 RabbitMQ 3.10.2 举例）：

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.10.2/rabbitmq_delayed_message_exchange-3.10.2.ez

执行安装命令：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

使用延迟队列

use function Workbunny\WebmanRabbitMQ\publish;

publish(new TestBuilder(), 'abc', headers: [
    'x-delay' => 10000, // 延迟 10 秒
]); // return bool

⚠️ 注意：向延迟队列发布普通消息会抛出一个 WebmanRabbitMQException 异常

替代方案

不少第三方厂商不支持安装延迟队列插件
当不支持安装延迟队列时，可以通过优先级队列 + REQUEUE 实现
- Builder 支持通过 REQUEUE 标记进行消息重入队尾
- 通过自定义 header 中的时间标记，和逻辑判断，当满足时间条件时则执行，不满足条件则通过 REQUEUE 将数据自动推回队尾
- 为了减少数据延迟问题，使用优先级标识将时间较近的消息优先级定义高一些，而时间较长的数据优先级定义低一些
  - 队列通常支持 0-9 的优先级，合理分配时间段和优先级的匹配关系

📤 生产消息

⚠️ 注意：

向延迟队列发布普通消息会抛出 WebmanRabbitMQPublishException 异常

首先使用命令行工具或者手动创建对应的 Builder，以下以 Workbunny\Tests\TestBuilders\TestPublishBuilder 举例

#	方式	适用场景	连接	Channel 生命周期
一	`publish($builder, $body)`	单次快捷发送	自动借还	回调模式，用完即还
二	`publish($builder, $body, connection: $conn)`	consumer 内复用连接	复用传入	回调模式，用完即还
三	`$builder->publish($connection, $config)`	需自定义 config	自行管理	回调模式，用完即还
四	`$channel->publish()`	完全自定义	自行管理	自行管理

💡 Tips
publish() 内部使用 channel(false, $callback) 回调模式借还 channel，不依赖 Coroutine::defer，HTTP 和 consumer 场景均可靠归还。

consumer 内消息流转：handler 中发消息时，建议复用 consumer 已有的连接，避免池借还竞态。
高频发送：批量或循环调用时，用 action() 外包一层，一次借还多次发送：
use function Workbunny\WebmanRabbitMQ\action;
$configs = [...]; // BuilderConfig[]
action(function (ConnectionInterface $connection) use ($configs) {
    $channel = $connection->channel();
    foreach ($configs as $config) {
        $channel->publish(
            $config->getBody(),
            $config->getHeaders(),
            $config->getExchange(),
            $config->getRoutingKey()
        );
    }
});

方式一：快捷发送

use function Workbunny\WebmanRabbitMQ\publish;
use Workbunny\Tests\TestBuilders\TestPublishBuilder;

publish(new TestPublishBuilder(), 'abc');

方式二：快捷复用发送

consumer handler 内复用已有的 $connection，避免每次从池子借还：

use function Workbunny\WebmanRabbitMQ\publish;

public function handler(BunnyMessage $message, Channel $channel, ConnectionInterface $connection): string
{
    // 复用 consumer 连接，不额外借还
    publish(new OtherBuilder(), $message->content, connection: $connection);

    return Constants::ACK;
}

方式三：构造发送

use Workbunny\Tests\TestBuilders\TestPublishBuilder;
use Workbunny\WebmanRabbitMQ\BuilderConfig;
use Workbunny\WebmanRabbitMQ\Connection\ConnectionInterface;

$builder = new TestPublishBuilder();

return $builder->action(function (ConnectionInterface $connection) use ($builder) {
    $config = new BuilderConfig($builder->getBuilderConfig()());
    $config->setBody('abc');
    return $builder->publish($connection, $config);
});

方式四：原生发送

使用 BuilderConfig 作为消息、配置的统一结构体：

use Workbunny\WebmanRabbitMQ\BuilderConfig;
use Workbunny\WebmanRabbitMQ\Connection\ConnectionInterface;
use Workbunny\WebmanRabbitMQ\ConnectionsManagement;

$config = new BuilderConfig();
$config->setExchange('your_exchange');
$config->setRoutingKey('your_routing_key');
$config->setQueue('your_queue');
$config->setBody('abc');
$config->setMandatory(true);
$config->setImmediate(false);

return ConnectionsManagement::connection(function (ConnectionInterface $connection) use ($config) {
    $connection->channel()->publish(
        $config->getBody(),
        $config->getHeaders(),
        $config->getExchange(),
        $config->getRoutingKey(),
        $config->getMandatory(),
        $config->getImmediate()
    );
});

📥 消费消息

⚠️ 注意：首先使用命令行工具或者手动创建对应的 Builder，以下以 Workbunny\Tests\TestBuilders\TestConsumeBuilder 举例

方式一：快捷消费

修改生成的 Builder 文件，将 handler() 方法逻辑添加消费逻辑
启动构建好的 Builder 自定义进程即是启动消费

方式二：Builder 消费

use Workbunny\Tests\TestBuilders\TestConsumeBuilder;
use Workbunny\WebmanRabbitMQ\Connection\ConnectionInterface;

$builder = new TestConsumeBuilder();
$builder->action(function (ConnectionInterface $connection) use ($builder) {
    $builder->consume($connection, $builder->getBuilderConfig());
});

⚠️ 注意：需要保持该进程常驻

🔀 消费与生产 TCP 隔离 `QueueSimpleBuilder`

为什么需要

PHP 的协程是单线程 + 协作式的：只有协程主动出让（suspend）才会切换。consumer 的 handler() 是在 onMessage 的协程中同步执行的（见 Channel::onBodyFramesComplete()），如果 handler() 内部使用同步阻塞 API（file_get_contents、curl_*、原生 PDO 等），整个事件循环会被占住：

handler() 阻塞 → 事件循环停摆 → 心跳定时器无法触发 → 服务端判定连接死亡 → 关闭连接

QueueBuilder 下消费与生产共用同一条连接（connections_pool.enable = false 时就是同一条 TCP），所以消费阻塞不仅会让消费者掉线，同进程内所有 producer 也会一起失效。

QueueSimpleBuilder 通过「消费独占一条连接、生产走另一条连接」把影响面隔离开：

┌── consumer 进程 ─────────────────────────────┐
│  Connection A ── TCP-1 ──┐  ← 消费专用（独占）│
│     ├ heartbeat          │                   │
│     └ channel: consume   │                   │
│                          │                   │
│  Connection B ── TCP-2 ──┤  ← 生产使用（池内）│
│     ├ heartbeat          ▼                   │
│     └ channel: publish  RabbitMQ             │
└──────────────────────────────────────────────┘

消费阻塞时只有 TCP-1 心跳失效，TCP-2 上的生产完全不受影响；消费者重连后消息仍在队列中继续消费。

⚠️ 前置要求

启动时会校验连接配置，不满足直接抛 WebmanRabbitMQException：

配置项	要求	原因
`connections_pool.enable`	必须为 `true`	池模式下才有多条连接可用
`connections_pool.min_connections`	`>= 2`	一条被消费长期占用，预留的另一条不能被 `idle_timeout` 回收
`connections_pool.max_connections`	`>= 2`	容量上限必须能容纳消费 + 生产两条连接

连接配置示例：

// config/plugin/workbunny/webman-rabbitmq/connections.php
return [
    'default' => [
        'connection'       => Connection::class,
        'connections_pool' => [
            'enable'          => true,   // ✅ 必须
            'min_connections' => 2,      // ✅ 必须 >= 2
            'max_connections' => 10,     // ✅ 必须 >= 2
            'idle_timeout'    => 60,
            'wait_timeout'    => 10,
        ],
        'config' => [
            // 消费者侧建议把心跳调大，给慢回调留出缓冲
            'channels_pool' => [
                'max_connections' => null,
                'idle_timeout'    => 60,
                'wait_timeout'    => 10,
            ],
        ],
    ],
];

创建 Builder

# 使用 simple 模式生成
php webman workbunny:rabbitmq-builder MySimpleBuilder -m simple

或手动继承：

<?php declare(strict_types=1);

namespace app\queue;

use Bunny\Message as BunnyMessage;
use Workbunny\WebmanRabbitMQ\Builders\QueueSimpleBuilder;
use Workbunny\WebmanRabbitMQ\Connection\Channel;
use Workbunny\WebmanRabbitMQ\Connection\ConnectionInterface;
use Workbunny\WebmanRabbitMQ\Constants;

class MySimpleBuilder extends QueueSimpleBuilder
{
    protected ?string $connection = 'default';

    protected array $queueConfig = [
        'name'           => 'my-queue',
        'prefetch_count' => 1,
    ];

    public function handler(BunnyMessage $message, Channel $channel, ConnectionInterface $connection): string
    {
        // 耗时逻辑（建议仍使用协程化 IO）
        $result = doSomething($message->content);

        // ✅ 发消息走连接池中另一条连接，与消费连接隔离
        $this->send(json_encode($result));

        return Constants::ACK;
    }
}

行为说明

环节	行为
`onWorkerStart`	先校验池配置 → `ConnectionsManagement::get()` 借出并长期持有一条连接（不 `release`）→ 在其上 `consume()`
`send()` / `publish($this, ...)`	走 `$this->action()` → 连接池借出另一条连接 → 用完归还
`onWorkerStop`	先释放持有的消费连接引用，再销毁整个连接池
池耗尽	记录 `channel pool exhausted` 日志 → 销毁连接池 → `Timer::sleep(restartInterval)` → `stopAll()`（由 webman 重新拉起进程）

使用注意

不要在 handler() 中把 consumer 连接传给 publish()：

// ❌ 会退回共用同一条 TCP，隔离失效
publish(new OtherBuilder(), 'body', connection: $connection);

// ✅ 让 Builder 自己从池中借另一条连接
publish(new OtherBuilder(), 'body');
$this->send('body');   // 等价写法（发往自身交换机/队列）

QueueSimpleBuilder 只解决「阻塞影响面隔离」，不解决阻塞本身：消费连接自身的心跳仍会因阻塞而超时，进程会掉线重连。要彻底避免，handler() 仍需使用协程化 IO （Coroutine\Http\Client、workerman/redis 等）。
每条连接都会维护独立的心跳与 Channel 池，连接数消耗高于 QueueBuilder；同一个 Builder 类不要同时用于高并发 HTTP 生产场景，生产侧请用普通 QueueBuilder。

workbunny / webman-rabbitmq

Package info

Security

Statistics

README

workbunny/webman-rabbitmq

📖 说明

✨ 简介

核心特性

🏗️ 架构概念

核心组件说明

🔨 Builder

🔗 Connection

📡 Channel

📦 AMQP

🚀 快速开始

系统要求

安装

⚙️ 配置

基础配置 `app.php`

连接配置 `connections.php`

🛠️ 命令行工具

⏰ 延迟队列

安装插件

使用延迟队列

替代方案

📤 生产消息

方式一：快捷发送

方式二：快捷复用发送

方式三：构造发送

方式四：原生发送

📥 消费消息

方式一：快捷消费

方式二：Builder 消费

🔀 消费与生产 TCP 隔离 `QueueSimpleBuilder`

为什么需要

⚠️ 前置要求

创建 Builder

行为说明

使用注意

workbunny / webman-rabbitmq

Package info

Security

Statistics

README

workbunny/webman-rabbitmq

📖 说明

✨ 简介

核心特性

🏗️ 架构概念

核心组件说明

🔨 Builder

🔗 Connection

📡 Channel

📦 AMQP

🚀 快速开始

系统要求

安装

⚙️ 配置

基础配置 app.php

连接配置 connections.php

🛠️ 命令行工具

⏰ 延迟队列

安装插件

使用延迟队列

替代方案

📤 生产消息

方式一：快捷发送

方式二：快捷复用发送

方式三：构造发送

方式四：原生发送

📥 消费消息

方式一：快捷消费

方式二：Builder 消费

🔀 消费与生产 TCP 隔离 QueueSimpleBuilder

为什么需要

⚠️ 前置要求

创建 Builder

行为说明

使用注意

基础配置 `app.php`

连接配置 `connections.php`

🔀 消费与生产 TCP 隔离 `QueueSimpleBuilder`