[PHP知识库] RabbitMQ基础【php版】

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> PHP知识库 -> RabbitMQ基础【php版】 -> 正文阅读

[PHP知识库]RabbitMQ基础【php版】

消息队列是一种异步的服务间通信方式，适用于无服务器和微服务架构。消息在被处理和删除之前一直存储在队列上。每条消息仅可被一位用户处理一次。消息队列可被用于分离重量级处理、缓冲或批处理工作以及缓解高峰期工作负载

特点

异步
消息队列本身是异步的，它允许接收者在消息发送很长时间后再取回消息

解耦
消息队列减少了服务之间的耦合性，不同的服务可以通过消息队列进行通信，而不用关心彼此的实现细节

广播

流量削峰与流控
当上下游系统处理能力存在差距的时候，利用消息队列做一个通用的 ”载体”。在下游有能力处理的时候，再进行分发与处理

RabbitMQ的基础结构图

RabbitMQ的相关概念

Message

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括 routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等

Publisher

消息的生产者，也是一个向交换器发布消息的客户端应用程序

Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列

Binding

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表

Queue

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走

Connection

网络连接，比如一个 TCP 连接。

Channel

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的 TCP 连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

Consumer

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序

Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 /

Broker

表示消息队列服务器实体。

以上是一些术语相关，掌握RabbitMQ，主要是要熟悉其6种工作模式，以及对应的使用场景

RabbitMQ六种工作模式

1. Work queues(工作队列模式)

流程示意图

在 Work queues 工作模式中，不需要设置交换器（RabbitMQ 会使用内部默认交换器进行消息转换），需要指定唯一的消息队列进行消息传递，并且可以有多个消息消费者。在这种模式下，多个消息消费者通过轮询的方式依次接收消息队列中存储的消息，一旦消息被某一个消费者接收，消息队列会将消息移除，而接收并处理消息的消费者必须在消费完一条消息后再准备接收下一条消息

特点

一条消息只会被一个消费端接收；
队列采用轮询的方式将消息是平均发送给消费者的；
消费者在处理完某条消息后，才会收到下一条消息。

代码示例

生产端new_task.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';

	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	// 从命令行中获取参数
	$data = implode(' ', array_slice($argv, 1));
	if (empty($data)) {
		$data = 'Hello World';
	}

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();

	// declare a queue
	$channel->queue_declare('hello', false, true, false, false);

	// create a message
	$msg = new AMQPMessage($data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

	// send the message to queue
	$channel->basic_publish($msg, '', 'hello');

	echo "[x] send ${data}!\n";

	$channel->close();

	try {
		$connection->close();
	} catch (Exception $e) {

	}
?>

消费端worker.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';

	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

	$channel = $connection->channel();
	$channel->basic_qos(null, 1, null);

	$channel->queue_declare('hello', false, false, false, false);


	echo "[*] Waiting for messages. \n";

	$callback = function (AMQPMessage $msg) {
		echo ' [x] Receive ', $msg->body, "\n";
		sleep(substr_count($msg->body, '.'));
		echo " [x] Done\n";
		$msg->ack();
	};

	$channel->basic_consume('hello', '', false, false, false, false, $callback);

	while ($channel->is_open()) {
		try {
			$channel->wait();
		} catch (ErrorException $e) {
		}
	}
?>

注意

在消费端，我们使用了message acknowledgments的方式，即设置channel的basic_consume方法的第4个参数为false(true means no ack),然后在回调函数callback中，使用$msg->ack()来发回ack消息。以避免因消费端因错误未能处理完消息，而rabbitMq确将消息从队列中删除，导致消息丢失的问题。ps: ack消息要从同样的channel中返回才行
如果忘记了发ack消息，后果会十分严重，因为rabbitMQ不会删除掉已经发出的消息，会导致吃掉越来越多的内存，我们可以使用命令来查询哪些消息没有收到ack确认
rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

如何确保消息不会因服务端的问题丢失？

声明队列为durable
$channel->queue_declare的第三个参数设置为true
标记我们的message为persistent
$msg = new AMQPMessage(
		$data,
		array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
);

在通常情况下，工作队列模式消费消息的方式叫做round-robin,会导致一些不公平情况发生。例如，在有两个工作人员的情况下，当所有的奇数消息都很重而偶数消息都很轻时，一个工作人员将一直很忙，而另一个几乎不做任何工作。那么应该如何解决？
使用channel的basic_qos方法，设置prefetch_count = 1, 这告诉RabbitMQ不要一次向worker提供多个消息。或者，换句话说，不要将新消息发送给worker，直到它已处理并确认前一个消息
$channel->basic_qos(null,1,null)

2. Publish/Subscribe (发布订阅模式）

流程示意图

在这里插入图片描述

在 Publish/Subscribe 工作模式中，必须先配置一个 fanout 类型的交换器，不需要指定对应的路由键（Routing key），同时会将消息路由到每一个消息队列上，然后每个消息队列都可以对相同的消息进行接收存储，进而由各自消息队列关联的消费者进行消费。

特点

每个消费者监听自己的队列；
生产者将消息发给 broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。

代码示例

生产端emit_log.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';

	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	// 建立连接
	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器
	$channel->exchange_declare('logs', 'fanout', false, false, false);

	// 从命令行中取出消息参数
	$data = implode(' ', array_slice($argv, 1));

	if (empty($data)) {
		$data = 'hello world';
	}

	// 生成消息
	$msg = new AMQPMessage($data);
	// 通过交换器发送消息
	$channel->basic_publish($msg, 'logs');

	echo ' [x] Sent ', $data, "\n";
	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}
?>

消费端receive_logs.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';

	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器，类型为fanout
	$channel->exchange_declare('logs', 'fanout', false, false, false);
	// 声明队列，不传入队列名，为随机队列
	// 第四个参数设置为true(exclusive),意味着当connection关闭，该队列也会删除
	[$queueName,,] = $channel->queue_declare('',false,false, true, false);
	// 将队列与交换器绑定
	$channel->queue_bind($queueName, 'logs');

	echo  " [*] Waiting for logs. To exit press CTRL+C\n";
	// 接收到消息后的回调函数
	$callback = function (AMQPMessage $msg) {
		echo ' [x] ', $msg->body, "\n";
	};

	$channel->basic_consume($queueName, '', false, true, false, false, $callback);

	while ($channel->is_open()) {
		try {
			$channel->wait();
		} catch (ErrorException $e) {
		}
	}
	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}

?>

运行测试

生产者

第一个消费者

第二个消费者

3. Routing (路由模式）

流程示意图

在这里插入图片描述

在 Routing 工作模式中，必须先配置一个 direct 类型的交换器，并指定不同的路由键值（Routing key）将对应的消息从交换器路由到不同的消息队列进行存储，由消费者进行各自消费

特点

每个消费者监听自己的队列，并且设置 routingkey；
生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列；

代码示例

生产端emit_log_direct.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Message\AMQPMessage;
	use PhpAmqpLib\Connection\AMQPStreamConnection;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器，类型为direct
	$channel->exchange_declare('direct_logs', 'direct', false, false, false);
	// 接收命令行参数（日志的级别）
	$severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';

	$data = implode(' ', array_slice($argv, 2));
	if (empty($data)) {
		$data = "Hello World!";
	}
	// 生成消息
	$msg = new AMQPMessage($data);
	// 发布消息
	$channel->basic_publish($msg, 'direct_logs', $severity);

	echo ' [x] Sent ', $severity, ':', $data, "\n";

	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}
?>

消费端receive_logs_direct.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Message\AMQPMessage;
	use PhpAmqpLib\Connection\AMQPStreamConnection;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器，类型为direct
	$channel->exchange_declare('direct_logs', 'direct', false, false, false);
	// 声明队列，不指定队列名，名字由系统随机生成返回到变量$queue_name
	[$queue_name, ,] = $channel->queue_declare("", false, false, true, false);

	$severities = array_slice($argv, 1);
	if (empty($severities)) {
		file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
		exit(1);
	}

	foreach ($severities as $severity) {
		// 绑定路由，这些路由规则，都会从队列中获取到消息
		$channel->queue_bind($queue_name, 'direct_logs', $severity);
	}

	echo " [*] Waiting for logs. To exit press CTRL+C\n";

	$callback = function (AMQPMessage $msg) {
		echo ' [x] ', $msg->getRoutingKey(), ':', $msg->body, "\n";
	};

	$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

	while ($channel->is_open()) {
		try {
			$channel->wait();
		} catch (ErrorException $e) {
		}
	}

	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}

?>

生产者产生 info 级别的 log

消费者1 接收 info warning类型的log

==消费者2 接收 warning error 类型的log (并未收到) ==

4. Topics（通配符模式）

流程示意图

在 Topics 工作模式中，必须先配置一个 topic 类型的交换器，并指定不同的路由键值（Routing key）将对应的消息从交换器路由到不同的消息队列进行存储，然后由消费者进行各自消费。Topics 模式与 Routing 模式的主要不同在于：Topics 模式设置的路由键是包含通配符的，其中，# 匹配多个字符，* 匹配一个字符，然后与其他字符一起使用 “.” 进行连接，从而组成动态路由键，在发送消息时可以根据需求设置不同的路由键从而将消息路由到不同的消息队列

特点

每个消费者监听自己的队列，并且设置带统配符的 routingkey
生产者将消息发给 broker，由交换机根据 routingkey 来转发消息到指定的队列

代码示例

生产端emit_log_topic.php

<?php

	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Message\AMQPMessage;
	use PhpAmqpLib\Connection\AMQPStreamConnection;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器，类型为topic
	$channel->exchange_declare('topic_logs', 'topic', false, false, false);
	// 路由参数由命令行参数传入
	$routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'anonymous.info';

	$data = implode(' ', array_slice($argv, 2));

	if (empty($data)) {
		$data = 'Hello World';
	}
	// message
	$msg = new AMQPMessage($data);
	// 发布消息
	$channel->basic_publish($msg, 'topic_logs', $routing_key);

	echo ' [x] Sent ', $routing_key, ':', $data, "\n";

	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}
?>

消费端receive_logs_topic.php

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明交换器，类型为topic
	$channel->exchange_declare('topic_logs', 'topic', false, false, false);
	[$queue_name, , ] = $channel->queue_declare("", false, false, true, false);

	$binding_keys = array_slice($argv, 1);
	if (empty($binding_keys)) {
		file_put_contents('php://stderr', "Usage: $argv[0] [binding_key]\n");
		exit(1);
	}
	// 依照传入的路由参数，将交换器与队列进行绑定
	foreach ($binding_keys as $binding_key) {
		$channel->queue_bind($queue_name, 'topic_logs', $binding_key);
	}

	echo " [*] Waiting for logs. To exit press CTRL+C\n";
	// 回调函数
	$callback = function (AMQPMessage $msg) {
		echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";
	};
	// 消费消息
	$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
	// 循环等待消息
	while ($channel->is_open()) {
		try {
			$channel->wait();
		} catch (ErrorException $e) {
		}
	}

	$channel->close();
	try {
		$connection->close();
	} catch (Exception $e) {
	}
?>

运行示例
消费端注册路由"kern.*"

消费端注册路由"*.critical"

生产端发送消息，使用路由"kern.info"

注册了"kern.*"路由的消费端收到了消息

5. RPC

流程示意图

RPC 工作模式与 Work queues 工作模式主体流程相似，都不需要设置交换器，需要指定唯一的消息队列进行消息传递。RPC 模式与 Work queues 模式的主要不同在于：RPC 模式是一个回环结构，主要针对分布式架构的消息传递业务，客户端 Client 先发送消息到消息队列，远程服务端 Server 获取消息，然后再写入另一个消息队列，向原始客户端 Client 响应消息处理结果。

特点

当客户端启动时，它会创建一个匿名的独家回调队列。
对于RPC请求，client将带有两个属性的消息：Reply_to，它被设置为回调队列和correlation_id，它被设置为每个请求的唯一值。
请求将发送到RPC_Queue队列。
RPC Worker（AKA：Server）正在等待该队列上的请求。出现请求时，使用Reply_to字段的队列，执行该作业并将结果与??结果发送回client。
client等待回调队列上的数据。出现邮件时，它会检查correlation_id属性。如果它与请求中的值匹配，则它将返回对应用程序的响应。

代码示例

生产端rpc_server.php(用于返回指定的fibonacci队列中的值)

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
	$channel = $connection->channel();
	// 声明队列，队列名为rpc_queue
	$channel->queue_declare('rpc_queue', false, false, false, false);
	// 处理：fib函数
	function fib ($n) {
		static $res = [];
		if (isset($res[$n])) {
			return $res[$n];
		}
		if ($n == 0) {
			return 0;
		}
		if ($n == 1) {
			return 1;
		}
		return fib($n-1) + fib($n-2);
	}

	echo " [x] Awaiting RPC requests\n";
	
	// 收到消息时的回调函数
	$callback = function (AMQPMessage $req) {
		$n = intval ($req->body);
		echo ' [.] fib(', $n , "\n";

		$msg = new AMQPMessage(
				(string) fib($n),
				array('correlation_id' => $req->get('correlation_id'))
			);
		// 服务端收到消息后通过同样的通道将返回消息传递给reply_to参数记录的队列
		$req->getChannel()->basic_publish(
				$msg,
				'',
				$req->get('reply_to')
			);
		// 发送ack消息告知此消息已处理
		$req->ack();
	};
	// 设置prefetch_count = 1, 这告诉RabbitMQ不要一次向server提供多个消息
	$channel->basic_qos(null, 1, null);
	// 从rpc_queue中取消息
	$channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);

	while ($channel->is_open()) {
		try {
			$channel->wait();
		} catch (ErrorException $e) {
		}
	}

	$channel->close();

	try {
		$connection->close();
	} catch (Exception $e) {
	}

?>

消费端rpc_client.php(请求fibonacci队列中的指定值)

<?php
	require_once dirname(__DIR__).'/vendor/autoload.php';
	use PhpAmqpLib\Connection\AMQPStreamConnection;
	use PhpAmqpLib\Message\AMQPMessage;

	class FibonacciRpcClient {
		private $connection;
		private $channel;
		private $callback_queue;
		private $response;
		private $corr_id;

		public function __construct() {
			$this->connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
			$this->channel = $this->connection->channel();
			// 随机声明一个队列
			list($this->callback_queue,,) = $this->channel->queue_declare('', false, false, true, false);
			// 从该队列中取出来自服务端的返回消息，并执行$this->onResponse回调
			$this->channel->basic_consume(
					$this->callback_queue,
					'',
					false,
					true,
					false,
					false,
					array($this, 'onResponse')
				);


		}
		// 回调函数，在回调函数中检验correlation_id
		public function onResponse(AMQPMessage $rsp) {
			if ($rsp->get('correlation_id') == $this->corr_id) {
				$this->response = $rsp->body;
			}
		}

		public function call($n) {
			$this->response = null;
			$this->corr_id = uniqid();
			// 向服务端发送的消息，消息中注明reply_to(回传的队列名),correlation_id(用于标记同一个请求响应)
			$msg = new AMQPMessage(
					(string) $n,
					array(
							'reply_to' => $this->callback_queue,
							'correlation_id' => $this->corr_id,
						)
				);
			// 将请求消息发送到rpc_queue中
			$this->channel->basic_publish($msg, '', 'rpc_queue');

			while (!$this->response) {
				try {
					$this->channel->wait();
				} catch (ErrorException $e) {
				}
			}
			return intval($this->response);
		}

	}
	$fibonacci_rpc = new FibonacciRpcClient();
	$response = $fibonacci_rpc->call(10);
	echo ' [.] Got ', $response, "\n";
?>