rabbitmq的实现（PHP--简单模式、fanout模式、direct模式、topic模式）

短命女 2023-02-13 11:22 1阅读 0赞

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    
    /**
     * @desc: 连接rabbitmq的一些操作，通用
     */
    
    class BasicConnect extends Controller
    {
    	public $mq_host 		= '192.168.XX.XXX';
    	public $mq_port1 		= 5672;
    	public $mq_port2 		= 5672;
    	public $mq_port3 		= 5673;
    	public $mq_user 		= 'XXXX';
    	public $mq_passwd 		= 'XXXX';
    	public $mq_vhost 		= '/';
    	
    	public $connection 		= '';
    	public function __construct(){
    		$host = [
    			[
    				'host'		=>	$this->mq_host, 
    				'port'		=>	$this->mq_port1, 
    				'user'		=>	$this->mq_user, 
    				'password'	=>	$this->mq_passwd, 
    				'vhost'		=>	$this->mq_vhost
    			],
    			[
    				'host'		=>	$this->mq_host, 
    				'port'		=>	$this->mq_port2, 
    				'user'		=>	$this->mq_user, 
    				'password'	=>	$this->mq_passwd, 
    				'vhost'		=>	$this->mq_vhost
    			],
    			[
    				'host'		=>	$this->mq_host, 
    				'port'		=>	$this->mq_port3, 
    				'user'		=>	$this->mq_user, 
    				'password'	=>	$this->mq_passwd, 
    				'vhost'		=>	$this->mq_vhost
    			],
    		];
    		
    		$options = [];
    		
    		$this->connection = AMQPStreamConnection::create_connection($host,$options);
    	}
    }

BasicConnect.php

SimpleProducer.php

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: 简单模式---消费者
     */
    
    class SimpleConsumer extends BasicConnect
    {
    	public $exchange 	= 'simple_test_exchange';
    	public $queue 		= 'simple_test_queue';
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    		
    		$channel = $this->connection->channel();
    		
    		/**
    		*	name: $queue    队列名称
    		*	passive: false  
    		*	durable: true  服务器重启后队列将仍存在（the queue will survive server restarts）
    		*	exclusive: false 可以通过其他通道访问队列（the queue can be accessed in other channels）
    		*	auto_delete: false 一旦通道关闭，队列将不会被删除。（the queue won't be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, true, false, false);
    
    		/**
    		*	name: $exchange
    		*	type: direct
    		*	passive: false
    		*	durable: true 服务器重启后交换机将仍存在(the exchange will survive server restarts)
    		*	auto_delete: false 一旦频道关闭，将不会删除交换。（the exchange won't be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::DIRECT, false, true, false);
    
    		$channel->queue_bind($this->queue, $this->exchange);
    
    		/**
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    		
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    }

SimpleConsumer.php

FanoutProducer.php

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: fanout形式--消费者1
     */
    
    class FanoutConsumer1 extends BasicConnect
    {
    	public $exchange 	= 'fanout_test_exchange';
    	public $queue 		= 'fanout_test_queue1';#注意就这个队列名不同
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    
    		$channel = $this->connection->channel();
    
    	   /**
    		*	name: $queue    队列名称，在fanout类型的交换机中必须唯一（should be unique in fanout exchange.）
    		*	passive: false  不检查是否存在同名队列（don't check if a queue with the same name exists）
    		*	durable: false  服务器重启后队列将无法生存（the queue will not survive server restarts）
    		*	exclusive: false 队列可能被其他通道访问（the queue might be accessed by other channels）
    		*	auto_delete: true 一旦通道关闭，队列将被删除（the queue will be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, false, false, true);
    
    	   /**
    		*	name: $exchange 交换机名称
    		*	type: fanout   交换机类型
    		*	passive: false 是否检查存在同名的交换机（don't check is an exchange with the same name exists）
    		*	durable: false 服务器重启后销毁交换机（the exchange won't survive server restarts）
    		*	auto_delete: true 管道（channel）关闭后将销毁交换机（the exchange will be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::FANOUT, false, false, true);
    
    		$channel->queue_bind($this->queue, $this->exchange);
    
    		
    		/*
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    
    }

FanoutConsumer1.php

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: fanout形式--消费者2
     */
    
    class FanoutConsumer2 extends BasicConnect
    {
    	public $exchange 	= 'fanout_test_exchange';
    	public $queue 		= 'fanout_test_queue2';#注意就这个队列名不同
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    
    		$channel = $this->connection->channel();
    
    	   /**
    		*	name: $queue    队列名称，在fanout类型的交换机中必须唯一（should be unique in fanout exchange.）
    		*	passive: false  不检查是否存在同名队列（don't check if a queue with the same name exists）
    		*	durable: false  服务器重启后队列将无法生存（the queue will not survive server restarts）
    		*	exclusive: false 队列可能被其他通道访问（the queue might be accessed by other channels）
    		*	auto_delete: true 一旦通道关闭，队列将被删除（the queue will be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, false, false, true);
    
    	   /**
    		*	name: $exchange 交换机名称
    		*	type: fanout   交换机类型
    		*	passive: false 是否检查存在同名的交换机（don't check is an exchange with the same name exists）
    		*	durable: false 服务器重启后销毁交换机（the exchange won't survive server restarts）
    		*	auto_delete: true 管道（channel）关闭后将销毁交换机（the exchange will be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::FANOUT, false, false, true);
    
    		$channel->queue_bind($this->queue, $this->exchange);
    
    		
    		/*
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    
    }

FanoutConsumer2.php

DirectProducer.php

[https://www.jianshu.com/p/fdb195698a5a][https_www.jianshu.com_p_fdb195698a5a]

[https://my.oschina.net/u/3698467/blog/1825316][https_my.oschina.net_u_3698467_blog_1825316]

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: topic形式--消费者1
     * @user: dingling
     * @version: V1.0
     * @date: 20200528
     */
    
    class DirectConsumer1 extends BasicConnect
    {
    	public $exchange 	= 'direct_test_exchange';
    	public $queue 		= 'direct_test_queue1';#注意就这个队列名不同
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    
    		$channel = $this->connection->channel();
    
    	   /**
    		*	name: $queue    队列名称，在fanout类型的交换机中必须唯一（should be unique in fanout exchange.）
    		*	passive: false  不检查是否存在同名队列（don't check if a queue with the same name exists）
    		*	durable: false  服务器重启后队列将无法生存（the queue will not survive server restarts）
    		*	exclusive: false 队列可能被其他通道访问（the queue might be accessed by other channels）
    		*	auto_delete: true 一旦通道关闭，队列将被删除（the queue will be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, false, false, true);
    
    	   /**
    		*	name: $exchange 交换机名称
    		*	type: DIRECT   交换机类型
    		*	passive: false 是否检查存在同名的交换机（don't check is an exchange with the same name exists）
    		*	durable: false 服务器重启后销毁交换机（the exchange won't survive server restarts）
    		*	auto_delete: true 管道（channel）关闭后将销毁交换机（the exchange will be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::DIRECT, false, false, true);
    
    		$channel->queue_bind($this->queue, $this->exchange, 'direct_key1');
    		
    		/*
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    
    }

DirectConsumer1.php

TopicProducer.php

\*（星号）可以代替一个任意标识符 ；\#（井号）可以代替零个或多个标识符。

[https://blog.csdn.net/a491857321/article/details/50616323][https_blog.csdn.net_a491857321_article_details_50616323]  （JAVA实现）

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: topic形式--消费者1
     */
    
    class TopicConsumer1 extends BasicConnect
    {
    	public $exchange 	= 'topic_test_exchange';
    	public $queue 		= 'topic_test_queue1';#注意就这个队列名不同
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    
    		$channel = $this->connection->channel();
    
    	   /**
    		*	name: $queue    队列名称，在fanout类型的交换机中必须唯一（should be unique in fanout exchange.）
    		*	passive: false  不检查是否存在同名队列（don't check if a queue with the same name exists）
    		*	durable: false  服务器重启后队列将无法生存（the queue will not survive server restarts）
    		*	exclusive: false 队列可能被其他通道访问（the queue might be accessed by other channels）
    		*	auto_delete: true 一旦通道关闭，队列将被删除（the queue will be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, false, false, true);
    
    	   /**
    		*	name: $exchange 交换机名称
    		*	type: TOPIC   交换机类型
    		*	passive: false 是否检查存在同名的交换机（don't check is an exchange with the same name exists）
    		*	durable: false 服务器重启后销毁交换机（the exchange won't survive server restarts）
    		*	auto_delete: true 管道（channel）关闭后将销毁交换机（the exchange will be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::TOPIC, false, false, true);
    
    		$channel->queue_bind($this->queue, $this->exchange, 'key.*');
    
    		
    		/*
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    
    }

TopicConsumer1.php

<?php
    namespace app\rabbitmq\controller;
    use think\Controller;
    use app\rabbitmq\controller\BasicConnect as BasicConnect;
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Exchange\AMQPExchangeType;
    /**
     * @desc: topic形式--消费者2
     */
    
    class TopicConsumer2 extends BasicConnect
    {
    	public $exchange 	= 'topic_test_exchange';
    	public $queue 		= 'topic_test_queue2';#注意就这个队列名不同
    	
    	public function __construct(){
    		parent::__construct();
    	}
    	
    	public function Index(){
    		$consumerTag = 'consumer' . getmypid();
    
    		$channel = $this->connection->channel();
    
    	   /**
    		*	name: $queue    队列名称，在fanout类型的交换机中必须唯一（should be unique in fanout exchange.）
    		*	passive: false  不检查是否存在同名队列（don't check if a queue with the same name exists）
    		*	durable: false  服务器重启后队列将无法生存（the queue will not survive server restarts）
    		*	exclusive: false 队列可能被其他通道访问（the queue might be accessed by other channels）
    		*	auto_delete: true 一旦通道关闭，队列将被删除（the queue will be deleted once the channel is closed.）
    		*/
    		$channel->queue_declare($this->queue, false, false, false, true);
    
    	   /**
    		*	name: $exchange 交换机名称
    		*	type: TOPIC   交换机类型
    		*	passive: false 是否检查存在同名的交换机（don't check is an exchange with the same name exists）
    		*	durable: false 服务器重启后销毁交换机（the exchange won't survive server restarts）
    		*	auto_delete: true 管道（channel）关闭后将销毁交换机（the exchange will be deleted once the channel is closed.）
    		*/
    		$channel->exchange_declare($this->exchange, AMQPExchangeType::TOPIC, false, false, true);
    
    		$channel->queue_bind($this->queue, $this->exchange, 'key.#');
    
    		
    		/*
    		*	queue: 队列名称，从何处获取消息的队列（Queue from where to get the messages）
    		*	consumer_tag: 消费者标识符（Consumer identifier）
    		*	no_local: 不接收此消费者发布的消息。（Don't receive messages published by this consumer.）
    		*	no_ack: 如果设置为真，则此用户将使用自动确认模式（If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.）
    		*	exclusive: 请求独占使用者访问，这意味着只有此使用者可以访问队列（Request exclusive consumer access, meaning only this consumer can access the queue）
    		*	nowait: 不要等待服务器响应。如果出现错误，服务器将引发通道异常（don't wait for a server response. In case of error the server will raise a channel exception）
    		*	callback: 回调函数（A PHP Callback）
    		*/
    
    		$channel->basic_consume($this->queue, $consumerTag, false, false, false, false, [new FanoutConsumer1(),'process_message']);
    
    		register_shutdown_function([new FanoutConsumer1(),'shutdown'], $channel, $this->connection);
    
    		// 循环，只要通道已注册回调(Loop as long as the channel has callbacks registered)
    		while ($channel->is_consuming()) {
    			$channel->wait();
    		}
    
    		exit;
    	}
    	
    	/**
    	 * 回调函数
    	 */
    	public function process_message($message){
    		echo "\n--------\n";
    		echo $message->body;
    		echo "\n--------\n";
    
    		$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    
    		// 发送字符串“quit”通知消费者退出。（Send a message with the string "quit" to cancel the consumer.）
    		if ($message->body === 'quit') {
    			$message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
    		}
    	}
    	
    	/**
    	 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
    	 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
    	 */
    	public function shutdown($channel, $connection)
    	{
    		$channel->close();
    		$connection->close();
    	}
    
    }

TopicConsumer2.php

[https_www.jianshu.com_p_fdb195698a5a]: https://www.jianshu.com/p/fdb195698a5a
[https_my.oschina.net_u_3698467_blog_1825316]: https://my.oschina.net/u/3698467/blog/1825316
[https_blog.csdn.net_a491857321_article_details_50616323]: https://blog.csdn.net/a491857321/article/details/50616323