feat(1.0): init

.gitignore

@@ -0,0 +1,2 @@
+/vendor/
+composer.lock

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 roiwk
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,185 @@
+# RabbitMQ
+rabbitmq async(workerman) and sync PHP Client, Producers, Consumers
+
+rabbitmq 是一个异步（workerman）和同步的PHP客户端，用于异步（workerman）和同步的生产者和消费者。
+
+
+# Dependencies 依赖
+
+php >= 8.0
+
+# Install 安装
+
+```sh
+composer require roiwk/rabbitmq
+```
+
+# Usage 使用
+
+[All demo](./example/)
+
+## Config Demo
+
+```php
+// 配置格式
+$config = [
+    'host' => '127.0.0.1',
+    'port' => 5672,
+    'vhost' => '/',
+    'mechanism' => 'AMQPLAIN',
+    'user' => 'username',
+    'password' => 'password',
+    'timeout' => 10,
+    'heartbeat' => 60,
+    'heartbeat_callback' => function(){},
+    'error_callback'     => null,
+];
+```
+
+## Publisher Demo
+```php
+// 同步发布者  sync Publisher
+Roiwk\Rabbitmq\Producer::connect($config)->publishSync('Hello World!', '', '', 'hello');
+
+// 异步发布者  async Publisher(workerman)
+
+use Workerman\Worker;
+
+$worker = new Worker();
+$worker->onWorkerStart = function() use($config) {
+    Roiwk\Rabbitmq\Producer::connect($config)->publishAsync('Hello World!', '', '', 'hello');
+};
+Worker::runAll();
+```
+
+
+## Consumer Demo
+
+```php
+// 同步消费者  sync Consumer
+
+use Bunny\AbstractClient;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+// style 1:
+$client = new Roiwk\Rabbitmq\Client($config, null, '', '', 'hello');
+$client->syncProcess(function(Message $message, Channel $channel, AbstractClient $client){
+    echo " [x] Received ", $message->content, "\n";
+    $channel->ack();
+});
+
+// style 2:
+$consumer = new class ($config) extends AbstractConsumer {
+    protected bool $async = false;
+    protected string $queue = 'hello';
+    protected array $consume = [
+        'noAck' => true,
+    ];
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+$consumer->onWorkerStart(null);
+
+```
+
+```php
+// 异步消费者  async Consumer(workerman)
+
+use Workerman\Worker;
+use Bunny\AbstractClient;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+$worker = new Worker();
+
+$consumer = new class ($config) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $queue = 'hello';
+
+    protected array $consume = [
+        'noAck' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+Worker::runAll();
+
+```
+
+# Advanced 高级用法
+
+## webman中自定义进程--消费者
+
+1.process.php
+```php
+'hello-rabbitmq' => [
+    'handler' => app\queue\rabbitmq\Hello::class,
+    'count'   => 1,
+    'constructor' => [
+        'rabbitmqConfig' => $config,
+        //'logger' => Log::channel('hello'),
+    ],
+]
+```
+2.app\queue\rabbitmq\Hello.php
+
+```php
+namespace app\queue\rabbitmq;
+
+use Roiwk\Rabbitmq\AbstractConsumer;
+use Roiwk\Rabbitmq\Producer;
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+
+class Hello extends AbstractConsumer
+{
+    protected bool $async = true;
+
+    protected string $queue = 'hello';
+
+    protected array $consume = [
+        'noAck' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+}
+
+```
+
+## webman中自定义进程--分组消费者
+  类似webman-queue插件, 分组将消费者放同一个文件夹下, 使用同一个worker, 多个进程数处理  
+1.process.php
+```php
+'hello-rabbitmq' => [
+    'handler' => Roiwk\Rabbitmq\GroupConsumers::class,
+    'count'   => 2,
+    'constructor' => [
+        'consumer_dir' => app_path().'/queue/rabbimq',
+        'rabbitmqConfig' => $config,
+        //'logger' => Log::channel('hello'),
+    ],
+]
+```
+2.在 ```app_path().'/queue/rabbimq'``` 目录下创建php文件, 继承```Roiwk\Rabbitmq\AbstractConsumer```即可, 同上```app\queue\rabbitmq\Hello.php```
+
+
+
+
+
+# Tips
+
+> 1. !!!  此库异步仅支持在workamn环境中, 同步环境都支持. 别的环境,如需异步, 请使用[bunny](https://packagist.org/packages/bunny/bunny)的客户端
+
+

composer.json

@@ -0,0 +1,32 @@
+{
+  "name": "roiwk/rabbitmq",
+  "type": "library",
+  "license": "MIT",
+  "description": "rabbitmq async(workerman) and sync Client, Producer, Consumer",
+  "keywords": [
+    "rabbitmq",
+    "amqp",
+    "queue",
+    "workerman",
+    "webman",
+    "plugin"
+  ],
+  "authors": [
+    {
+      "name": "roiwk",
+      "email": "vip@roiwk.cn"
+    }
+  ],
+  "minimum-stability": "stable",
+  "require": {
+    "php": "^8.0",
+    "workerman/rabbitmq": "^1.0",
+    "psr/log": "^1.0||^2.0||^3.0",
+    "illuminate/collections": "^8.0|^9.0|^10.0"
+  },
+  "autoload": {
+    "psr-4": {
+      "Roiwk\\Rabbitmq\\": "src"
+    }
+  }
+}

examples/1-hello-world/README.md

@@ -0,0 +1,14 @@
+# 1 Hello World
+[http://www.rabbitmq.com/tutorials/tutorial-one-php.html](http://www.rabbitmq.com/tutorials/tutorial-one-php.html)
+
+```
+php receive-async.php start
+```
+
+```
+php send-async.php start
+```
+
+```
+php send.php
+```

examples/1-hello-world/receive-async.php

@@ -0,0 +1,40 @@
+<?php
+
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+$consumer = new class ($config, $log) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $queue = 'hello';
+
+    protected array $consume = [
+        'noAck' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+
+
+Worker::runAll();

examples/1-hello-world/send-async.php

@@ -0,0 +1,25 @@
+<?php
+use Monolog\Logger;
+use Roiwk\Rabbitmq\Producer;
+use Bunny\Channel;
+use Bunny\Message;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker->onWorkerStart = function()  {
+
+    $config = require __DIR__ . '/../config.php';
+    $log = require __DIR__ . '/../log.php';
+
+    Producer::connect($config, $log)->publishAsync('Hello World!', '', '', 'hello');
+
+};
+Worker::runAll();

examples/1-hello-world/send.php

@@ -0,0 +1,19 @@
+<?php
+
+
+use Roiwk\Rabbitmq\Producer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+Producer::connect($config, $log)->publishSync('Hello World!', '', '', 'hello');
+
+echo " [x] Sent 'Hello World!'\n";
+

examples/2-work-queues/README.md

@@ -0,0 +1,25 @@
+# 2 Work Queues
+[http://www.rabbitmq.com/tutorials/tutorial-two-php.html](http://www.rabbitmq.com/tutorials/tutorial-two-php.html)
+
+```
+php worker-async.php start
+```
+
+
+```
+php new_task-asnyc.php start First message.
+php new_task-asnyc.php start Second message..
+php new_task-asnyc.php start Third message...
+php new_task-asnyc.php start Fourth message....
+php new_task-asnyc.php start Fifth message.....
+
+```
+
+```
+php new_task.php First message.
+php new_task.php Second message..
+php new_task.php Third message...
+php new_task.php Fourth message....
+php new_task.php Fifth message.....
+
+```

examples/2-work-queues/new_task-async.php

@@ -0,0 +1,28 @@
+<?php
+
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker->onWorkerStart = function($worker) {
+
+    $config = require __DIR__ . '/../config.php';
+    $log = require __DIR__ . '/../log.php';
+
+    global $argv;
+    unset($argv[1]);
+    $data = implode(' ', array_slice($argv, 1));
+
+    Producer::connect($config, $log)->publishAsync($data, '', '', 'task_queue', [], ['delivery-mode' => 2]);
+};
+
+
+Worker::runAll();

examples/2-work-queues/new_task.php

@@ -0,0 +1,22 @@
+<?php
+
+
+use Roiwk\Rabbitmq\Producer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+
+$data = implode(' ', array_slice($argv, 1));
+
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+Producer::connect($config, $log)->publishSync($data, '', '', 'task_queue', [], ['delivery-mode' => 2]);
+
+echo " [x] Sent '{$data}'\n";
+

examples/2-work-queues/worker-async.php

@@ -0,0 +1,45 @@
+<?php
+
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker = new Worker();
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+$consumer = new class ($config, $log) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $queue = 'task_queue';
+
+    protected array $qos = [
+        'prefetch_size' => 0,
+        'prefetch_count' => 1,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+        sleep(substr_count($message->content, '.'));
+        echo " [x] Done", $message->content, "\n";
+        $channel->ack($message);
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+
+Worker::runAll();

examples/3-publish-subscribe/README.md

@@ -0,0 +1,13 @@
+# 3 Publish/Subscribe
+[http://www.rabbitmq.com/tutorials/tutorial-three-php.html](http://www.rabbitmq.com/tutorials/tutorial-three-php.html)
+
+
+
+```
+php receive_logs-async.php start
+```
+
+```
+php emit_log-async.php start
+```
+

examples/3-publish-subscribe/emit_log-async.php

@@ -0,0 +1,30 @@
+<?php
+use Bunny\Channel;
+use Bunny\Message;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker->onWorkerStart = function() {
+    global $argv;
+    unset($argv[1]);
+    $data = implode(' ', array_slice($argv, 1));
+    if (empty($data)) {
+        $data = "info: Hello World!";
+    }
+
+    $config = require __DIR__ . '/../config.php';
+    $log = require __DIR__ . '/../log.php';
+
+    Producer::connect($config, $log)->publishAsync($data, 'logs', 'fanout');
+
+};
+Worker::runAll();

examples/3-publish-subscribe/emit_log.php

@@ -0,0 +1,20 @@
+<?php
+
+use Bunny\Client;
+use Roiwk\Rabbitmq\Producer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+$data = 'Hello World!';
+
+Producer::connect($config, $log)->publishSync($data, 'logs', 'fanout');
+
+echo " [x] Sent '{$data}'\n";
+

examples/3-publish-subscribe/receive_logs-async.php

@@ -0,0 +1,44 @@
+<?php
+
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+$consumer = new class ($config, $log) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $exchange = 'logs';
+
+    protected string $exchangeType = 'fanout';
+
+    protected string $queue = 'log_queue';
+
+    protected array $consume = [
+        'noAck' => true,
+        'noLocal' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+
+Worker::runAll();

examples/4-routing/README.md

@@ -0,0 +1,10 @@
+# 4 Routing
+[http://www.rabbitmq.com/tutorials/tutorial-four-php.html](http://www.rabbitmq.com/tutorials/tutorial-four-php.html)
+
+```
+php receive_logs-async.php start info warning error
+```
+
+```
+php emit_log-async.php start
+```

examples/4-routing/emit_log-async.php

@@ -0,0 +1,32 @@
+<?php
+use Bunny\Channel;
+use Bunny\Message;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker->onWorkerStart = function() {
+    global $argv;
+    unset($argv[1]);
+    $argv = array_values($argv);
+    $severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
+    $data = implode(' ', array_slice($argv, 2));
+    if (empty($data)) {
+        $data = "Hello World!";
+    }
+
+    $config = require __DIR__ . '/../config.php';
+    $log = require __DIR__ . '/../log.php';
+
+    Producer::connect($config, $log)->publishAsync($data, 'direct_logs', 'direct', $severity);
+};
+
+Worker::runAll();

examples/4-routing/emit_log.php

@@ -0,0 +1,22 @@
+<?php
+
+use Roiwk\Rabbitmq\Producer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
+$data = implode(' ', array_slice($argv, 2));
+if (empty($data)) {
+    $data = "Hello World!";
+}
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+Producer::connect($config, $log)->publishSync($data, 'direct_logs', 'direct', $severity);
+echo " [x] Sent ",$severity,':',$data," \n";
+

examples/4-routing/receive_logs-async.php

@@ -0,0 +1,47 @@
+<?php
+
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+
+$consumer = new class ($config, $log) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $exchange = 'direct_logs';
+
+    protected string $exchangeType = 'direct';
+
+    protected string $queue = 'direct_logs_queue';
+
+    protected array $routingKeys = ['info', 'warning', 'error'];
+
+    protected array $consume = [
+        'noAck' => true,
+        'noLocal' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+
+Worker::runAll();

examples/5-topics/README.md

@@ -0,0 +1,14 @@
+# 5 Topics
+[http://www.rabbitmq.com/tutorials/tutorial-five-php.html](http://www.rabbitmq.com/tutorials/tutorial-five-php.html)
+
+
+```
+php receive_logs_topic-async.php start "#"
+php receive_logs_topic-async.php start "kern.*"
+php receive_logs_topic-async.php start "*.critical"
+php receive_logs_topic-async.php start "kern.*" "*.critical"
+```
+
+```
+php emit_log_topic-async.php start
+```

examples/5-topics/emit_log_topic-async.php

@@ -0,0 +1,32 @@
+<?php
+use Bunny\Channel;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$worker->onWorkerStart = function() {
+    global $argv;
+    unset($argv[1]);
+    $argv = array_values($argv);
+    $routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
+    $data = implode(' ', array_slice($argv, 2));
+    if (empty($data)) {
+        $data = "Hello World!";
+    }
+
+    $config = require __DIR__ . '/../config.php';
+    $log = require __DIR__ . '/../log.php';
+
+    Producer::connect($config, $log)->publishAsync($data, 'topic_logs', 'topic', $routing_key);
+
+};
+
+Worker::runAll();

examples/5-topics/emit_log_topic.php

@@ -0,0 +1,23 @@
+<?php
+
+use Roiwk\Rabbitmq\Producer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
+$data = implode(' ', array_slice($argv, 2));
+if (empty($data)) {
+    $data = "Hello World!";
+}
+
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+Producer::connect($config, $log)->publishSync($data, 'topic_logs', 'topic', $routing_key);
+echo " [x] Sent ",$routing_key,':',$data," \n";
+

examples/5-topics/receive_logs_topic-async.php

@@ -0,0 +1,51 @@
+<?php
+
+use Bunny\Channel;
+use Bunny\Message;
+use Bunny\AbstractClient;
+use Workerman\Worker;
+use Roiwk\Rabbitmq\Producer;
+use Roiwk\Rabbitmq\AbstractConsumer;
+
+if (file_exists(__DIR__ . '/../../../../../vendor/autoload.php')) {
+    require __DIR__ . '/../../../../../vendor/autoload.php';
+} else {
+    require __DIR__ . '/../../vendor/autoload.php';
+}
+
+$worker = new Worker();
+
+$config = require __DIR__ . '/../config.php';
+$log = require __DIR__ . '/../log.php';
+
+$consumer = new class ($config, $log) extends AbstractConsumer {
+
+    protected bool $async = true;
+
+    protected string $exchange = 'topic_logs';
+
+    protected string $exchangeType = 'topic';
+
+    protected string $queue = 'topic_logs_queue';
+
+    protected array $routingKeys = [
+        '#',
+        'kern.*',
+        '*.critical',
+        'kern.*" "*.critical',
+    ];
+
+    protected array $consume = [
+        'noAck' => true,
+        'noLocal' => true,
+    ];
+
+    public function consume(Message $message, Channel $channel, AbstractClient $client)
+    {
+        echo " [x] Received ", $message->content, "\n";
+    }
+};
+
+$worker->onWorkerStart = [$consumer, 'onWorkerStart'];
+
+Worker::runAll();

examples/config.php

@@ -0,0 +1,14 @@
+<?php
+
+return [
+    'host' => '127.0.0.1',
+    'port' => 5672,
+    'vhost' => '/',
+    'mechanism' => 'AMQPLAIN',
+    'user' => 'admin',
+    'password' => '123456',
+    'timeout' => 10,
+    'heartbeat' => 60,
+    'heartbeat_callback' => function(){},
+    'error_callback' => null,
+];

examples/log.php

@@ -0,0 +1,9 @@
+<?php
+
+if (class_exists(Monolog\Logger::class)) {
+    $log = new Monolog\Logger('test');
+    $log->pushHandler(new Monolog\Handler\StreamHandler('php://stdout'));
+    return $log;
+} else {
+    return null;
+}

src/AbstractConsumer.php

@@ -0,0 +1,107 @@
+<?php
+
+namespace Roiwk\Rabbitmq;
+
+use Psr\Log\LoggerInterface;
+
+abstract class AbstractConsumer implements Consumable
+{
+    protected string $exchange = '';
+    protected string $exchangeType = '';
+
+    protected string $queue = '';
+
+    // topic exchange - routingKeys
+    protected array $routingKeys = [];
+
+    protected array $exchangeDeclareDefault = [
+        'passive' => false,
+        'durable' => true,
+        'auto_delete' => false,
+        'internal' => false,
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $queueDeclareDefault = [
+        'passive' => false,
+        'durable' => true,
+        'auto_delete' => false,
+        'exclusive' => false,
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $queueBindDefault = [
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $consumeDefault = [
+        'consumerTag' => '',
+        'noLocal' => false,
+        'noAck' => false,
+        'exclusive' => false,
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $qosDefault = [
+        'prefetch_size' => 0,
+        'prefetch_count' => 1,
+    ];
+
+    protected array $exchangeDeclare = [];
+    protected array $queueDeclare = [];
+    protected array $queueBind = [];
+    protected array $consume = [];
+    protected array $qos = [];
+
+    protected $client;
+    protected bool $async = true;
+
+    public function __construct(
+        protected array $rabbitmqConfig,
+        protected ?LoggerInterface $logger = null,
+    ){
+        $this->init();
+    }
+
+    public function init()
+    {
+        $initProperty = [
+            'exchangeDeclare' => 'exchangeDeclareDefault',
+            'queueDeclare' => 'queueDeclareDefault',
+            'queueBind' => 'queueBindDefault',
+            'consume' => 'consumeDefault',
+            'qos' => 'qosDefault',
+        ];
+
+        array_walk($initProperty, function ($default, $current) {
+            if (empty($this->{$current})) {
+                $this->{$current} = $this->{$default};
+            } else {
+                $this->{$current} = array_replace_recursive($this->{$default}, $this->{$current});
+            }
+        });
+
+        $this->client = new Client(
+            $this->rabbitmqConfig, $this->logger, $this->exchange, $this->exchangeType,
+            $this->queue, $this->routingKeys, $this->exchangeDeclare, $this->queueDeclare,
+            $this->queueBind, $this->consume, $this->qos
+        );
+    }
+
+    public function onWorkerStart($worker): void
+    {
+        if (is_a(static::class, AbstractConsumer::class, true) || is_subclass_of(static::class, Consumable::class)) {
+            if ($this->async) {
+                $this->client->asyncProcess([$this, 'consume']);
+            } else {
+                $this->client->syncProcess([$this, 'consume']);
+            }
+        } else {
+            return;
+        }
+    }
+}

src/Client.php

@@ -0,0 +1,185 @@
+<?php
+
+namespace Roiwk\Rabbitmq;
+
+use Bunny\Channel;
+use Bunny\Message;
+use Illuminate\Support\Arr;
+use Psr\Log\LoggerInterface;
+use Workerman\RabbitMQ\Client as AsyncClient;
+
+class Client
+{
+    public function __construct(
+        protected array $config,
+        protected ?LoggerInterface $logger = null,
+        protected string $exchange = '',
+        protected string $exchangeType = '',
+        protected string $queue = '',
+        protected array $routingKeys = [],
+        protected array $exchangeDeclare = [],
+        protected array $queueDeclare = [],
+        protected array $queueBind = [],
+        protected array $consume = [],
+        protected array $qos = [])
+    {
+    }
+
+    public function getErrorCallback(): \Closure
+    {
+        if (!empty($this->config['error_callback'])) {
+            return $this->config['error_callback'];
+        } else {
+            return function (\Throwable $throwable) {
+                $this->logger?->error('['.getmypid().']:'.$throwable->getMessage().PHP_EOL, [$throwable->getTraceAsString()]);
+            };
+        }
+    }
+
+    public function asyncDeclare(Channel $channel)
+    {
+        if (empty($this->exchangeType)) {
+            return $channel->queueDeclare(
+                $this->queue, $this->queueDeclare['passive'],
+                $this->queueDeclare['durable'], $this->queueDeclare['exclusive'], $this->queueDeclare['auto_delete'],
+                $this->queueDeclare['nowait'], $this->queueDeclare['arguments']
+            )
+            ->then(function () use ($channel) {
+                return $channel;
+            });
+        } else {
+            return $channel->exchangeDeclare($this->exchange, $this->exchangeType,
+                $this->exchangeDeclare['passive'], $this->exchangeDeclare['durable'],
+                $this->exchangeDeclare['auto_delete'], $this->exchangeDeclare['internal'],
+                $this->exchangeDeclare['nowait'], $this->exchangeDeclare['arguments']
+            )
+                ->then(function () use ($channel) {
+                    return $channel->queueDeclare($this->queue, $this->queueDeclare['passive'],
+                        $this->queueDeclare['durable'], $this->queueDeclare['exclusive'], $this->queueDeclare['auto_delete'],
+                        $this->queueDeclare['nowait'], $this->queueDeclare['arguments']
+                    )
+                        ->then(function () use ($channel) {
+                            $promises = [];
+
+                            if (empty($this->routingKeys)) {
+                                $promises[] = $channel->queueBind($this->queue, $this->exchange, '', $this->queueBind['nowait'], $this->queueBind['arguments']);
+                            } else {
+                                foreach ($this->routingKeys as $binding_key) {
+                                    $promises[] = $channel->queueBind($this->queue, $this->exchange, $binding_key, $this->queueBind['nowait'], $this->queueBind['arguments']);
+                                }
+                            }
+
+                            return \React\Promise\all($promises)->then(function () use ($channel) {
+                                return $channel;
+                            });
+                        });
+                });
+        }
+    }
+
+    public function asyncProcess(callable $consumer)
+    {
+        $reject = $this->getErrorCallback();
+
+        (new AsyncClient($this->config, $this->logger))->connect()
+            ->then(function (AsyncClient $client) {
+                return $client->channel();
+            }, $reject)
+            ->then(function (Channel $channel) {
+                return $channel->qos($this->qos['prefetch_size'] ?? 0, $this->qos['prefetch_count'] ?? 0)
+                    ->then(function () use ($channel) {
+                        return $this->asyncDeclare($channel)->then(function () use ($channel) {
+                            return $channel;
+                        });
+                    });
+            }, $reject)
+            ->then(function (Channel $channel) use ($reject, $consumer) {
+                $this->logger?->debug('Waiting:['.getmypid().'] Waiting for messages.', []);
+                $channel->consume(
+                    function (Message $message, Channel $channel, AsyncClient $client) use ($reject, $consumer) {
+                        $this->logger?->info('Received:['.getmypid().']: '.$message->content, [$message]);
+
+                        try {
+                            call_user_func_array($consumer, [$message, $channel, $client]);
+                        } catch (\Throwable $throw) {
+                            if (!$this->consume['noAck']) {
+                                $channel->nack($message);
+                            }
+                            $reject($throw);
+                        }
+
+                        return;
+                    },
+                    $this->queue,
+                    $this->consume['consumerTag'],
+                    $this->consume['noLocal'],
+                    $this->consume['noAck'],
+                    $this->consume['exclusive'],
+                    $this->consume['nowait'],
+                    $this->consume['arguments'],
+                );
+            }, $reject);
+    }
+
+    public function syncDeclare(Channel $channel)
+    {
+        if (empty($this->exchangeType)) {
+            $channel->queueDeclare(
+                $this->queue, $this->queueDeclare['passive'],
+                $this->queueDeclare['durable'], $this->queueDeclare['exclusive'], $this->queueDeclare['auto_delete'],
+                $this->queueDeclare['nowait'], $this->queueDeclare['arguments']
+            );
+        } else {
+            $channel->exchangeDeclare($this->exchange, $this->exchangeType,
+                $this->exchangeDeclare['passive'], $this->exchangeDeclare['durable'],
+                $this->exchangeDeclare['auto_delete'], $this->exchangeDeclare['internal'],
+                $this->exchangeDeclare['nowait'], $this->exchangeDeclare['arguments']
+            );
+            $channel->queueDeclare($this->queue, $this->queueDeclare['passive'],
+                $this->queueDeclare['durable'], $this->queueDeclare['exclusive'], $this->queueDeclare['auto_delete'],
+                $this->queueDeclare['nowait'], $this->queueDeclare['arguments']
+            );
+            if (empty($this->routingKeys)) {
+                $channel->queueBind($this->queue, $this->exchange, '', $this->queueBind['nowait'], $this->queueBind['arguments']);
+            } else {
+                foreach ($this->routingKeys as $binding_key) {
+                    $channel->queueBind($this->queue, $this->exchange, $binding_key, $this->queueBind['nowait'], $this->queueBind['arguments']);
+                }
+            }
+        }
+    }
+
+    public function syncProcess(callable $consumer)
+    {
+        try {
+            $reject = $this->getErrorCallback();
+            $rabbitmqConfig = Arr::only($this->config, ['host', 'port', 'vhost', 'user', 'password']);
+            $client = (new \Bunny\Client($rabbitmqConfig))->connect();
+            $channel = $client->channel();
+            $this->syncDeclare($channel);
+            $channel->consume(
+                function (Message $message, Channel $channel, \Bunny\Client $client) use ($consumer) {
+                    $this->logger?->info('Received:['.getmypid().']: '.$message->content, [$message]);
+                    try {
+                        call_user_func_array($consumer, [$message, $channel, $client]);
+                    } catch (\Throwable $throw) {
+                        if (!$this->consume['noAck']) {
+                            $channel->nack($message);
+                        }
+                        throw $throw;
+                    }
+                },
+                $this->queue,
+                $this->consume['consumerTag'],
+                $this->consume['noLocal'],
+                $this->consume['noAck'],
+                $this->consume['exclusive'],
+                $this->consume['nowait'],
+                $this->consume['arguments'],
+            );
+            $client->run();
+        } catch (\Throwable $throwable) {
+            $reject($throwable);
+        }
+    }
+}

src/Consumable.php

@@ -0,0 +1,12 @@
+<?php
+
+namespace Roiwk\Rabbitmq;
+
+use Bunny\AbstractClient;
+use Bunny\Channel;
+use Bunny\Message;
+
+interface Consumable
+{
+    public function consume(Message $message, Channel $channel, AbstractClient $client);
+}

src/GroupConsumers.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Roiwk\Rabbitmq;
+
+use Psr\Log\LoggerInterface;
+use support\Container;
+
+/**
+ * 分组消费.(模仿webman-redis queue).
+ */
+class GroupConsumers
+{
+    public function __construct(
+        protected $consumer_dir = '',
+        protected array $rabbitmqConfig,
+        protected ?LoggerInterface $logger = null,
+    ) {
+    }
+
+    public function onWorkerStart($worker): void
+    {
+        if (!is_dir($this->consumer_dir)) {
+            echo "Consumer directory {$this->consumer_dir} not exists\r\n";
+
+            return;
+        }
+        $dir_iterator = new \RecursiveDirectoryIterator($this->consumer_dir);
+        $iterator = new \RecursiveIteratorIterator($dir_iterator);
+        foreach ($iterator as $file) {
+            if (is_dir($file)) {
+                continue;
+            }
+            $fileinfo = new \SplFileInfo($file);
+            $ext = $fileinfo->getExtension();
+            if ('php' === $ext) {
+                $class = str_replace('/', '\\', substr(substr($file, strlen(base_path())), 0, -4));
+                if (!is_a($class, AbstractConsumer::class, true)) {
+                    continue;
+                }
+
+                $consumer = Container::make($class, [$this->rabbitmqConfig, $this->logger]);
+                $consumer->onWorkerStart($worker);
+            }
+        }
+    }
+}

src/Producer.php

@@ -0,0 +1,130 @@
+<?php
+
+namespace Roiwk\Rabbitmq;
+
+use Bunny\Channel;
+use Bunny\Client;
+use Illuminate\Support\Arr;
+use Psr\Log\LoggerInterface;
+use Workerman\RabbitMQ\Client as AsyncClient;
+
+class Producer
+{
+    protected array $exchangeDeclareDefault = [
+        'passive' => false,
+        'durable' => true,
+        'auto_delete' => false,
+        'internal' => false,
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $queueDeclareDefault = [
+        'passive' => false,
+        'durable' => true,
+        'exclusive' => false,
+        'auto_delete' => false,
+        'nowait' => false,
+        'arguments' => [],
+    ];
+
+    protected array $exchangeDeclare = [];
+
+    protected array $queueDeclare = [];
+
+    public function __construct(
+        protected array $rabbitmqConfig,
+        protected ?LoggerInterface $logger = null)
+    {
+    }
+
+    public static function connect(array $rabbitmqConfig, LoggerInterface $logger = null)
+    {
+        return new self($rabbitmqConfig, $logger);
+    }
+
+    protected function setDeclare(array $exchangeOrQueueDeclare, string $exchange = '', string $exchangeType = '')
+    {
+        if (!empty($exchange) && !empty($exchangeType)) {
+            $this->exchangeDeclare = array_replace_recursive($this->exchangeDeclareDefault, $exchangeOrQueueDeclare);
+        } else {
+            $this->queueDeclare = array_replace_recursive($this->queueDeclareDefault, $exchangeOrQueueDeclare);
+        }
+    }
+
+    protected function declare(Channel $channel, string $routingOrQueue, string $exchange = '', string $exchangeType = '')
+    {
+        if (!empty($exchange) && !empty($exchangeType)) {
+            return $channel->exchangeDeclare($exchange, $exchangeType,
+                $this->exchangeDeclare['passive'], $this->exchangeDeclare['durable'],
+                $this->exchangeDeclare['auto_delete'], $this->exchangeDeclare['internal'],
+                $this->exchangeDeclare['nowait'], $this->exchangeDeclare['arguments']);
+        } else {
+            return $channel->queueDeclare($routingOrQueue,
+                $this->queueDeclare['passive'], $this->queueDeclare['durable'],
+                $this->queueDeclare['exclusive'], $this->queueDeclare['auto_delete'],
+                $this->queueDeclare['nowait'], $this->queueDeclare['arguments']);
+        }
+    }
+
+    public function publishAsync(string $data, string $exchange = '', string $exchangeType = '', string $routingOrQueue = '',
+        array $exchangeOrQueueDeclare = [], array $headers = [], bool $mandatory = false, bool $immediate = false
+    ) {
+        $this->setDeclare($exchangeOrQueueDeclare, $exchange, $exchangeType);
+
+        $reject = function (\Throwable $throwable) {
+            $this->logger?->error('['.getmypid().']:'.$throwable->getMessage().PHP_EOL.$throwable->getTraceAsString(), [__CLASS__]);
+        };
+        (new AsyncClient($this->rabbitmqConfig, $this->logger))->connect()
+            ->then(function (AsyncClient $client) {
+                return $client->channel();
+            }, $reject)
+            ->then(function (Channel $channel) use ($exchange, $exchangeType, $routingOrQueue) {
+                return $this->declare($channel, $routingOrQueue, $exchange, $exchangeType)
+                    ->then(function () use ($channel) {
+                        return $channel;
+                    });
+            }, $reject)
+            ->then(function (Channel $channel) use ($exchange, $routingOrQueue, $data, $headers, $mandatory, $immediate) {
+                $this->logger?->info('('.getmygid().') Sending :'.$data, [__CLASS__]);
+
+                return $channel->publish($data, $headers, $exchange, $routingOrQueue, $mandatory, $immediate)
+                    ->then(function () use ($channel) {
+                        return $channel;
+                    });
+            }, $reject)
+            ->then(function (Channel $channel) use ($data) {
+                $this->logger?->info('('.getmygid().') Sent :'.$data, [__CLASS__]);
+
+                $client = $channel->getClient();
+
+                return $channel->close()->then(function () use ($client) {
+                    return $client;
+                });
+            }, $reject)
+            ->then(function (AsyncClient $client) {
+                $client->disconnect();
+            }, $reject);
+    }
+
+    public function publishSync(string $data, string $exchange = '', string $exchangeType = '', string $routingOrQueue = '',
+        array $exchangeOrQueueDeclare = [], array $headers = [], bool $mandatory = false, bool $immediate = false
+    ) {
+        $this->setDeclare($exchangeOrQueueDeclare, $exchange, $exchangeType);
+
+        $rabbitmqConfig = Arr::only($this->rabbitmqConfig, ['host', 'port', 'vhost', 'user', 'password']);
+
+        try {
+            $client = (new Client($rabbitmqConfig))->connect();
+            $channel = $client->channel();
+            $this->declare($channel, $routingOrQueue, $exchange, $exchangeType);
+            $published = $channel->publish($data, $headers, $exchange, $routingOrQueue, $mandatory, $immediate);
+            $channel->close();
+            $client->disconnect();
+        } catch (\Throwable $throwable) {
+            $this->logger?->error('['.getmypid().']:'.$throwable->getMessage().PHP_EOL.$throwable->getTraceAsString(), [__CLASS__]);
+        }
+
+        return $published ?? false;
+    }
+}