消息队列(MQ)，很多场景都有它的身影，MQ的主要功能包括应用解耦、流量削峰、异步处理。本文主要讲解RabbitMq的原理及应用实例，将参考官网文档重点介绍RabbitMq基本概念、work queue模式、fanout模式、direct模式、topic模式、RPC实现、publisher confirms机制，从而达到快速入门的目的。
0.RabbitMq基本概念
- vhost，虚拟主机，提供了完全隔离独立的环境，包括exchange、queue等，可通过插件web管理后台或者rabbitmqctl命令设置user的vhost权限。
- connection，要使用rabbitmq必然要与服务器建立连接了，AMQP协议是基于TCP连接的应用层协议。
- channel，信道用于复用connection，减少TCP连接带来的资源开销，当访问量大的时候则需要开辟多个connection，并分摊到chennel。
- routing_key，路由键在pub/sub模式下作为exchange匹配binding到queue的条件；在work queue模式下，可视为队列名称发送消息。
- exchange，交换机在信道内，负责接受并转发消息。根据交换机的类型，有不同的匹配方式。
- binding_key，绑定值可视为exchange与queue之间的映射关系值，绑定值与queue之间的关系是n:n，当一个queue对应exchange的多个binding_key时，exchange只会发送一次到该queue。
- queue，消息队列。
- message，消息是要传递及处理的数据，通过RabbitMq指定的类来构造，可配置消息的参数属性，如correlation_id（请求标识），delivery_mode（投递模式）等。
- producer/publisher，消息的生产者/发布者，携带routing_key和msg。
- consumer/subscriber，消息的消费者/订阅者，按照不同的模式处理队列中的消息。
1.work queues模式
1.work queues
常规的消息队列模式，不涉及交换机exchange和队列绑定queue_binding，执行过程：生产者发送消息至队列，消费者从队列中取数据消费。
producer代码示例（PHP）
//1.建立连接
$connection = new AMQPStreamConnection('localhost', 5672, 'root', 'root');
//2.信道
$channel = $connection->channel();
//3.信道中声明队列
$queue_name='task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
$message = "Hello Task";
//4.生成amqp消息
$msg = new AMQPMessage($message, [
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
]);//投递模式设置为消息持久化
//5.发布消息
$channel->basic_publish($msg, '', $queue_name);
echo "publisher Sent '{$message}!'\n";
$channel->close();
$connection->close();
consumer代码示例
$connection = new AMQPStreamConnection('localhost', 5672, 'root', 'root');
$channel = $connection->channel();
$queue_name='task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo "consumer received : " . $msg->body . PHP_EOL;
sleep(1);
echo "Done" . PHP_EOL;
//确认消息
$msg->ack();
};
//公平调度, 设置预加载个数
$channel->basic_qos(null, 1, null);
//持续监听，回调处理消息
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
while ($channel->is_open()) {
$channel->wait();
}
下面介绍publish/subscribe模式，并引入exchange和queue_binding。该模式根据exchange的不同类型有不同的转发规则，exchange的类型主要有fanout、direct、topic。
2.fanout模式
2.fanout
该模式引入exchange、queue_binding，但不涉及routing_key和binding_key，因为publisher把消息投递给exchange后，所有绑定在该交换机上的队列都能接收到消息。
publisher代码
...
//通用连接部分参考上面，后面代码同理，只展示核心变更部分;完整代码可看官网
//该模式不用声明队列，只需声明exchange
$channel->exchange_declare('fanout_logs', 'fanout', false, true, false);//1.fanout交换机
..
//消息投递到交换机
$channel->basic_publish($msg, 'fanout_logs');//2.fanout模式
subscriber代码
...
$channel->exchange_declare('fanout_logs', 'fanout', false, true, false);//1.声明交换机
...
$channel->queue_bind($queue_name, 'fanout_logs');//2.队列绑定交换机
...
比起work queue，该模式更灵活，利用exchange可将消息转发到多个queue中。
3.direct模式
3.direct
如果在pub/sub模式下，只想将交换机的消息转发给指定的队列，fanout模式显然无法满足。此时可以利用direct模式，该模式将exchange和queue通过binding_key绑定在一起；exchange在接收publisher消息时依据routing_key和binding_key是否完全匹配，决定是否转发到对应queue。
publisher代码
$channel->exchange_declare('direct_logs', 'direct', false, true, false);//1.direct交换机
$routing_key = 'black';
$channel->basic_publish($msg, 'direct_logs', $routing_key);//2.发布消息至交换机，携带routing_key
subscriber代码
...
$channel->exchange_declare('direct_logs', 'direct', false, true, false);
$bindingKey = 'black';
$channel->queue_bind($queue_name, 'direct_logs', $bindingKey);//队列绑定交换机,声明binding_key
...
4.topic模式
4.topic
topic模式在direct模式基础上升级，routing_key和binding_key非完全匹配，支持更灵活的匹配规则；routing_key/binding_key可以通过word1.word2.wordn方式进行灵活扩展。【符号*代表1个word，符号#可代表0或n个words】
publisher代码
$channel->exchange_declare('topic_logs', 'topic', false, false, false); //3.1.topics路由
$routing_key = 'black.tall.big';
$channel->basic_publish($msg, 'topic_logs', $routing_key);//2.发布消息至交换机，携带routing_key
subscriber代码
$channel->exchange_declare('topic_logs', 'topic', false, true, false);//topic模式
$bindingKey = '#';//相当于全部消息都能接收
$channel->queue_bind($queue_name, 'topic_logs', $bindingKey);//队列绑定交换机,声明binding_key
bindingKey的举🌰
成功：black.#，自动匹配2个words、'black.tall.*'匹配1个word，占位匹配时必须要有点号.
失败：black.short.*
失败-错误使用符号：black#
5.RPC模式
5.RPC
RPC, 全称remote procedure call即远程程序调用，比起常规的远程调用，基于RabbitMq的RPC优点有：1.异步调用；2.方便扩展提升服务端性能(开启多个server)
5.1.实现原理�
- 服务端和客户端，通过两个队列进行通信，RPC队列rpc_queue和回调队列reply_to_queue。
- 客户端携带请求标识correlation_id和reply_to_queue回调队列信息，发送请求至rpc_queue，服务端监听rpc_queue，消费消息并发送消息至指定回调队列reply_to_queue。
- 客户端监听回调队列reply_to_queue并通过correlation_id获取请求处理结果。
下面以计算斐波那契数为作为RPC示例。
client端代码
class FibonacciRpcClient
{
private $connection;
private $channel;
private $callback_queue;
private $response;
private $corr_id;
//构造函数，监听回调队列，处理
public function __construct()
{
$this->connection = new AMQPStreamConnection(
'localhost',
5672,
'root',
'root'
);
$this->channel = $this->connection->channel();
//1.生成回调队列
$this->callback_queue = 'reply_to';
$this->channel->queue_declare($this->callback_queue, false, true, false, false);
//2.1.轮训消费
$this->channel->basic_consume(
$this->callback_queue,
'',
false,
true,
false,
false,
array(
$this,
'onResponse'
)
);
}
//2.1.2监听队列的回调函数
public function onResponse($rep)
{
if ($rep->get('correlation_id') == $this->corr_id) {
$this->response = $rep->body;
}
}
//远程调用,发送消息至rpc队列
public function call($n)
{
$this->response = null;
$this->corr_id = uniqid();//3.生成请求的唯一标识
//4.1.创建消息，携带请求标识、回调队列名称
$msg = new AMQPMessage(
(string)$n,
array(
'correlation_id' => $this->corr_id,
'reply_to' => $this->callback_queue
)
);
//4.2.发送消息至rpc队列，等待服务端消费
$this->channel->basic_publish($msg, '', 'rpc_queue');
//5.循环判断结果
while (!$this->response) {
$this->channel->wait();
}
return intval($this->response);
}
}
$fibonacci_rpc = new FibonacciRpcClient();//构造函数，监听回调队列reply_to
$response = $fibonacci_rpc->call(35);//发送消息至prc队列，并循环判断回调队列的处理结果。
echo ' [.] Got ', $response, "\n";//回调队列的处理结果
server端代码
$connection = new AMQPStreamConnection('localhost', 5672, 'root', 'root');
$channel = $connection->channel();
//声明队列
$channel->queue_declare('rpc_queue', false, false, false, false);
function fib($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 ($req) {
//1.1监听rpc队列，处理client发送的消息
$n = intval($req->body);
echo ' [.] fib(', $n, ")\n";
//1.2.返回处理结果，并携带请求标识
$msg = new AMQPMessage(
(string) fib($n),
array('correlation_id' => $req->get('correlation_id'))
);
//2.发送消息至同一信道的 回调队列， 由client监听消费。
$req->delivery_info['channel']->basic_publish(
$msg,
'',
$req->get('reply_to')
);
//3.消息接受确认
$req->ack();
};
//设置预加载数量，服务端worker公平调度
$channel->basic_qos(null, 1, null);
//轮训消费，监听rpc队列
$channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);
while ($channel->is_open()) {
$channel->wait();
}
$channel->close();
$connection->close();
调用结果
client:
[.] Got 9227465
server:
[x] Awaiting RPC requests
[.] fib(35)
6.publisher confirms模式
publisher confirms是RabbitMq实现可靠传输的扩展，用来判断publisher是否成功把消息发送到RabbitMq的broker。RabbitMq实现可靠传输的方式有两种：事务（不推荐）、publisher confirms，这两种方式互斥。publisher confirms的实现方式又可分为：同步、异步。
- 6.1. 同步实现
该模式是基于信道的，所以只要增加两个步骤即可：
6.1.1. 信道声明为confirm模式
6.1.2. 声明同步等待的超时时间
代码如下：
...
$channel->confirm_select();//1.声明信道为confirm模式
...
try {
$channel->wait_for_pending_acks($timeOut);//2.同步等待timeOut时间
}catch (Exception $exception){
echo "exception:" . $exception->getMessage() . PHP_EOL;
}
..
- 6.2. 异步实现
异步实现通过注册回调的两个方法set_ack_handler和set_nack_handler。
代码如下
$channel->confirm_select();//1.声明信道为confirm模式
//2.消息被ack后的回调
$channel->set_ack_handler(function (AMQPMessage $msg) {
echo "ack msg" . PHP_EOL;
file_put_contents('./ackfile.txt',json_encode($msg),FILE_APPEND);
});
//3.消息被nack'ed后的回调
$channel->set_nack_handler(function (AMQPMessage $msg) {
echo "nack msg" . PHP_EOL;
file_put_contents('./nackfile.txt',json_encode($msg),FILE_APPEND);
});
$channel->wait_for_pending_acks();
以上只是RabbitMq各种模式的基本使用，其他很多特性(持久化、网络分区、集群等)并未涉及，若要使用更多的特性请查阅官网文档，然后手动跑一下代码才能理解得更好。希望本文能帮助大家对RabbitMq的使用有个大致了解。
