rabbitmq -- 消息的可靠性
rabbitmq作为我们系统之间沟通的桥梁，消息的可靠性就显得格外重要。
假如rabbitmq crash掉之后重新启动，原本的交换机、队列、消息都会消失，如果我们队列中存在一些很重要的消息的时候，我们并不愿意这样的事情发生，这就需要借助rabbitmq的持久化机制。
rabbitmq的持久化
交换机、队列设置可以为持久化，消息的投递模式也可以选择持久化（三个持久化）。
持久化的消息在进入持久化队列之后会写入到rabbitmq的持久性日志文件中，消息被消费掉后，会把持久性日志文件中该消息标记为等待垃圾收集
在broker重启之后，持久化的交换机、队列会重新初始化，持久化的消息会投递到原先的队列上
怎么做到消息持久化
三个持久化：
交换机持久化、队列持久化、消息持久化
代码示例：
<?php
try{
$config = [
"host" => "127.0.0.1",
"port" => "5672",
"login" => "wuj13",
"password" => "Aaa294515505",
"vhost" => "test_vhost"
];
$conn = new AMQPConnection($config);
//连接rabbitmq
if(!$conn->connect()){
throw new Exception("连接失败");
}
//创建信道
$channel = new AMQPChannel($conn);
//创建交换机
$exchange = new AMQPExchange($channel);
$exchangeName = "exchange_one";
//设置交换机名称
$exchange->setName($exchangeName);
//设置交换机类型
$exchange->setType(AMQP_EX_TYPE_DIRECT);
//设置为持久化交换机
$exchange->setFlags(AMQP_DURABLE);
//生成交换机
$exchange->declareExchange();
//创建队列
$queueName ="queue_test";
$queue = new AMQPQueue($channel);
//设置队列名称
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
//生成队列
$queue->declareQueue();
//队列绑定交换机
$queue->bind($exchangeName,$queueName);
$message = "hello world";
//向交换机发布消息
$re = $exchange->publish($message,$queueName,AMQP_NOPARAM,["delivery_mode" => 2]);
if($re){
echo "成功发送消息".$message.PHP_EOL;
}
else{
throw new Exception("发送消息失败");
}
}
catch(Exception $exception){
echo "操作异常 ".$exception->getMessage();
}
finally{
$channel->close();
$conn->disconnect();
}
其中
$exchange->setFlags(AMQP_DURABLE)设置交换机为durable代表持久类型
$queue->setFlags(AMQP_DURABLE)设置队列为持久类型
$exchange->publish($message,$queueName,AMQP_NOPARAM,["delivery_mode" => 2])向交换机发送消息时设置消息的delivery_mode为2代表消息是持久类型的
代码运行后，我们到管理界面去看看，发现有一条持久化的消息
image.png
我们把rabbitmq服务给重启一下，交换机、队列、消息，全部都还在，代表我们的消息持久化已经基本做好了，大家可以自己去试一试。
但我们不妨想一想，如果我们的broker在把消息写入日志文件的时候crash掉了，那我们的消息不就丢了吗？对的，其实这是有很小的概率会出现这样的情况，这个时候可以借助mirror queue来解决这个问题，这个我们后面会专门讲这个东西，不然篇幅就太长了。
那我们先认为很大程度上已经解决了rabbitmq这一端因为宕机而导致消息丢失的问题，但是我们还面临着两个问题：
1 我们的程序怎么保证我们的消息能够成功到达队列
2 怎么保证我们的消息消费成功的
我们一个个来解决！
我们先看一幅图
image.png
消息要到达队列中并处以稳定状态（已持久化），要经过三个流程
1 消息到达交换机
2 交换机把消息路由到队列中
3 消息持久化写入日志
第一个流程可能由于网络的波动、延迟等原因，消息没有准确达到交换机
第二个流程可能routing key错误等原因，消息没有准确到达队列
第三个流程可能在消息写入日志的过程中，broker crash掉了，重新后消息丢失。
也就是说，在没有其他措施的情况下，客户端发送消息这个操作，消息如果想要在队列中处于稳定地状态，要经过三个独立的环节，并不是一个原子操作，三个环节中的任意一个环节出错，都有可能导致消息丢失，而我们的生产者对此一无所知，会影响我们业务的正常进行。
看到原子化，我们首先想到用事务来解决这个问题，但是在rabiitmq中，事务会很大幅度地降低性能，我们来看一个对比数据：
在同样开启持久化的情况下，没有开启事务，写入10万条数据仅需11秒
image.png
微信截图_20201128222508.png
而在同样的条件下，我们模拟10万次事务的独立开启提交，写入10万条数据需要208秒！
image.png
微信截图_20201128222108.png
这相差了接近20倍的性能，所以一般情况下我们不会选择事务来解决消息发送可能失败的问题，我们选择"发送者确认模式"来解决这个问题。
发送者确认模式
我们可以通过$channel->confirmSelect() 来开启信道的发送者确认模式，开启之后，信道会为每一条消息分配一个唯一的消息id，即delivery_tag，从0开始，依次是1,2,3这样，每个信道之间的delivery_tag是相互独立的。
我们先来了解两个东西：ack 、nack
ack ：
在发送者确认模式中，有两种情况服务端会给客户端发送ack确认信号
1 消息到达交换机，交换机发现该消息可以路由到队列，便将消息路由到队列，并向客户端发送一个ack信号（注意：消息如果是持久化的，需要等持久化完成后，才会向客户端发送ack信号。如果是开启了镜像队列，需要所有镜像中的队列收到消息后，才会发送ack信号），即消息成功入列且稳定后会向客户端发送ack信号
2 消息到达交换机，交换机发现根据消息的routing key，无法将该消息路由到任何队列，就会向客户端发送一个ack信号。如果发送消息时，我们指定了mandatory参数为true,指定该消息为强制到达，那么在向客户端发送ack信号之前，还有将该消息返回给客户端，顺序是 "消息" ->ack
nack :
消息到达交换机，交换机路由消息的过程中，由于系统内部错误而无法继续下去的时候，就会向客户端发送一个nack信号，告诉客户端，这个消息处理不了。
经过以上的阐述，我们要怎么知道这条消息到底有没有成功到达队列并处于稳定呢？
我们发现
ack无法作为消息是否成功入列的判断标准！！！
ack无法作为消息是否成功入列的判断标准！！！
ack无法作为消息是否成功入列的判断标准！！！
重要的事情说三遍
消息成功入列会给客户端发ack信号，消息无法路由到队列也会给客户端发ack信号，但是这两者是有区别的。成功入列的话是单纯发送"ack"，而如果消息无法路由到队列则是发送 "消息" + "ack",且"消息"比ack信号先到
那我们的程序中拿什么来作为消息是否成功入列的判断标准呢?相信各位看客心中已经有了答案。
没错，如果返回了"消息",那么自然是失败的。如果没有返回消息体，单纯返回ack ，那就是成功的。废话不多说，上代码
<?php
try{
$config = [
"host" => "127.0.0.1",
"port" => "5672",
"login" => "wuj13",
"password" => "Aaa294515505",
"vhost" => "test_vhost"
];
$conn = new AMQPConnection($config);
//连接rabbitmq
if(!$conn->connect()){
throw new Exception("连接失败");
}
//创建信道
$channel = new AMQPChannel($conn);
//开启消息确认模式
$channel->confirmSelect();
$isSuccess = true;
//设置回调方法
$channel->setConfirmCallback(function($delivery_tag, $multiple) use ($channel,$conn,&$isSuccess){
echo "收到ack回执：\n";
echo "delivery_tag: $delivery_tag \n";
echo "multiple:$multiple\n";
if($isSuccess){
echo "消息成功入列\n";
/**
* TODO
* 1 数据库事务提交
* 2 业务处理完成，返回成功
*/
}
else{
echo "消息入列失败\n";
/**
* TODO
* 1 数据库事务回滚
* 2 失败消息记录日志
* 3 返回失败
*/
}
$channel->close();
$conn->disconnect();
exit;
},function($delivery_tag, $multiple,$requeue){
echo "收到nack回执：delivery_tag: ".$delivery_tag. " multiple: $multiple"." requeue:".$requeue.PHP_EOL;
/**
* TODO
* 重新发送消息
*/
});
//设置消息返还的回调
$channel->setReturnCallback(function($reply_code, $reply_text, $exchange, $routing_key, $properties, $body) use(&$isSuccess){
$isSuccess = false;
//进行日志记录
echo "收到return消息\n";
echo "body: $body \n";
echo "replay_code: $reply_code \n";
echo "replay_text: $reply_text \n";
});
//创建交换机
$exchange = new AMQPExchange($channel);
$exchangeName = "exchange_one";
//设置交换机名称
$exchange->setName($exchangeName);
//设置交换机类型
$exchange->setType(AMQP_EX_TYPE_DIRECT);
//设置为持久化交换机
$exchange->setFlags(AMQP_DURABLE);
//生成交换机
$exchange->declareExchange();
// //创建队列
$queueName ="queue_test";
$queue = new AMQPQueue($channel);
//设置队列名称
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
//生成队列
$queue->declareQueue();
//队列绑定交换机
$queue->bind($exchangeName,$queueName);
$message = "hello world";
for($i = 1; $i<=1;$i++){
$exchange->publish("message".$i,"123",AMQP_MANDATORY,["delivery_mode" => 2]);
}
$channel->waitForConfirm();
}
catch(Exception $exception){
echo "操作异常 ".$exception->getMessage();
$channel->close();
$conn->disconnect();
}
代码解析：
$channel->confirmSelect() 我们使用confirmSelect来开启信道的发送者确认模式（注意，信道一旦开启发送者确认模式，就无法使用事务，这两者不兼容）
$channel->setConfirmCallback() 是设置回调，参数一是ack回调，服务端返回ack信号时触发。参数二是nack回调，服务端返回nack信号时触发
$channel->setReturnCallback() 是设置消息返还的回调，服务端返回"消息"时触发
$channel->waitForConfirm() 是进入阻塞模式等待服务端返回通知
至于在各个回调中应该怎么去处理业务，请注意看我代码中的注释！
先说说nack回调的处理
nack回调是属于rabbitmq系统内部错误时返回的信号，这种情况属于很少见，这个时候我们应该把消息进行重新发送
而ack回调跟return回调我们要结合起来，用来判断消息有没有成功入列
如果触发了return回调，代表消息必然路由失败，这时我们就把$isSuccess变量设置为false，便于后续到达的ack信号触发ack回调的时候进行成功与否的判断
ack回调中，如果发现$isSuccess为false，证明在ack信号达到前，已经触发过return回调了，证明消息是入列失败的，否则证明单纯只有ack信号达到，则是成功入列的。
如果消息入列失败，我们需要回滚我们的数据库事务，并把失败的消息记录日志，给前端返回错误。此时不会重新发送该消息，因为即使我们重新发送消息，也是无法路由的。
如果入列成功，那就提交数据库事务，然后给前端正常返回业务数据即可。
然后就是断开信道、rabbitmq的连接，中止阻塞进程。
