webman-mq

基于 Webman 的单机高性能消息队列插件：内存队列 + WAL 分片 Broker，无需 Redis。

仓库：https://github.com/lax1024git/webman-mq

特性

• Direct / Topic / Fanout 交换机
• 多 Broker 分片（TCP / Unix Socket）
• WAL + checkpoint + CRC；大消息 Blob 外置
• 延时投递 / 失败重试 / 死信
• 业务用法对齐 webman/redis-queue：Client::send / Consumer

要求

• PHP >= 8.1
• Webman ^2.1

安装

composer require lax1024git/webman-mq

安装后会复制配置到：

• config/plugin/webman/mq/app.php
• config/plugin/webman/mq/process.php

并自动创建（若不存在）：

• app/queue/mq/ExampleWebmanMqConsumer.php

按需修改 exchanges / queue_shards，并在 .env 中调整：

QUEUE_SHARD_COUNT=4
QUEUE_USE_UNIX_SOCKET=false
QUEUE_CONSUMER_COUNT=8
QUEUE_CONFIRM_DEFAULT=fsync_batch
QUEUE_BROKER_MEMORY_LIMIT=512M

重启 Webman：

php start.php restart
# Windows: php windows.php

快速使用

发布：

\Webman\Mq\Client::send('order_queue', ['id' => 1]);
\Webman\Mq\Client::publish('webmanmq_exchange', 'webmanmq.create', ['id' => 1]);

// 或
\Webman\Mq\MqQueue::send('order_queue', ['id' => 1]);

消费（app/queue/mq/YourConsumer.php）：

namespace app\queue\mq;

use Webman\Mq\Consumer;

class YourConsumer implements Consumer
{
    public string $queue = 'order_queue';

    public function consume($data): void
    {
        // 必须幂等；正常返回 → ACK；抛异常 → 延时重试
    }
}

运维

php webman mq:stats
php webman mq:health
php webman mq:clear order_queue --status=all

定位说明

当前定位为单机消息队列（at-least-once，业务侧需幂等）。默认 fsync_batch 会在确认前对 WAL 做真实 fsync。

License

MIT