发表新帖

Swoole 协程如何使用rabbitmq 多信道？

```php
Co\run(function () {

Co::set(['hook_flags' => SWOOLE_HOOK_FILE]);
				$connection = new AMQPStreamConnection(env('RABBITMQ_HOST') , env('RABBITMQ_PORT') , env('RABBITMQ_LOGIN') , env('RABBITMQ_PASSWORD') , env('RABBITMQ_VHOST') , false , 'AMQPLAIN' , null , 'en_US' , 3 , 30 , null , true , 15);
				if (!$connection->isConnected()) {
						throw new \Exception('Connecte Fail.');
				}
				$wg = new WaitGroup();
				for ($i = 1 ; $i <= 4 ; $i++) {
						$wg->add(1);
						Co::create(function () use ($connection,$i,$wg) {
								stdDump($i);
								$channel = $connection->channel();
								defer(function () use ($channel,$wg) {
										$channel->close();
										$wg->done();
								});
								$exChanegName = env('RABBITMQ_EXCHANGE');
								$channel->exchange_declare($exChanegName , 'fanout' , false , true , false);
								$queue_name = env('RABBITMQ_QUEUE' , "serve_toc_fs");
								$channel->queue_declare($queue_name , false , true , false , false , false , new AMQPTable([
										'x-message-ttl'   => 3000 ,
										'durable'         => true ,
										'consumer_no_ack' => true ,
										'timeout'         => 0 ,
										'prefetch_count'  => 0 , // 限流
										'persistent'      => true , // required if you want to listen forever
								]));
								$channel->queue_bind($queue_name , $exChanegName);
								$callback = function ($msg)use ($i) {
										stdDump("{$i}->",$msg->body);
										Co::sleep(1);
										return;
								};
								$channel->basic_consume($queue_name , "TAG_{$i}" , false , true , false , false , $callback);
								while ($channel->is_consuming()) {
										$channel->wait();
								}
						});
				}
				$wg->wait();

});
```

查看 Connections Channels 有4个 ，2个闲置的 未绑定到的队列
查看rabbitmq Queue Consumers 只有 2个 channel

发布于1年前 · 28 次浏览 · 来自提问