上篇我们用redis实现了每5分钟重试一次帖子提交失败的重试，俗话说，狡兔三窟，啊不，要学会举一反三，现在我们用rabbitmq来实现一下之前的操作
首先来安装rabbitmq，我是在wsl上安装的，你可以根据自己的操作系统选择安装。
执行如下命令：
// 安装rabbitmq服务 sudo apt install rabbitmq-server // 启动rabbitmq服务 sudo service rabbitmq-server start // 安装浏览器界面插件 sudo rabbitmq-plugins enable rabbitmq_management // 添加用户 sudo rabbitmqctl add_user bcm 123456 // 把用户设置为超管 sudo rabbitmqctl set_user_tags bcm administrator // 给用户添加权限 configure write read '.'：无此权限 '.*'：有此权限 sudo rabbitmqctl set_permissions -p / bcm '.*' '.*' '.*'
其他命令：
// 查询用户 sudo rabbitmqctl list_users // 查询用户权限 sudo rabbitmqctl list_permissions
一切准备就绪，你就可以用浏览器打开 http://127.0.0.1:15672了，用刚才配置的账号密码登录，如果不配置账号密码，可以用默认的guest guest 登录。
此时我写了代码尝试连接，却连接失败了，需要执行如下命令
sudo vim /etc/rabbitmq/rabbitmq-env.conf // 在文件中添加如下命令 NODE_IP_ADDRESS=127.0.0.1 NODE_PORT=5672
至此，就可以开心的写代码测试了。
实现思路：如果提交失败，放入延时队列中，设置过期时间为5分钟，此延时队列不做消费处理，过期后，进入死信队列，监听死信队列，消费，如果不成功继续放入延时队列，直到成功或到达重试上限次数。
代码展示
private $host = "127.0.0.1";
private $port = 5672;
private $user = 'bcm';
private $password = '123456';
protected $connection;
protected $channel;
public function __construct(){
$this->connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password);
$this->channel = $this->connection->channel();
}
public function createExchange($exchangeName, $type, $passive = false, $durable = false, $autoDelete = false){
$this->channel->exchange_declare($exchangeName, $type, $passive, $durable, $autoDelete);
}
public function createQueue($queueName, $pasive = false, $durable = false, $exlusive = false, $autoDelete = false, $nowait = false, $arguments = []){
$this->channel->queue_declare($queueName, $pasive, $durable, $exlusive, $autoDelete, $nowait, $arguments);
}
public function sendMessage($message, $routeKey, $exchange = '', $properties = []){
$data = new AMQPMessage(
$message, $properties
);
$this->channel->basic_publish($data, $exchange, $routeKey);
}
public function consumeMessage($queueName, $callback, $tag = '', $noLocal = false, $noAck = false, $exclusive = false, $noWait = false){
$this->channel->basic_consume($queueName, $tag, $noLocal, $noAck, $exclusive, $noWait, $callback);
while ($this->channel->is_consuming()) {
try {
$this->channel->wait();
} catch (ErrorException $e) {
echo $e->getMessage();
}
}
}
public function __destruct(){
$this->channel->close();
try {
$this->connection->close();
} catch (Exception $e) {
echo $e->getMessage();
}
}
}
实现延时队列
public function createQueue($ttl, $deadExName, $deadQueueName, $queueName)
{
$args = new AMQPTable([
'x-dead-letter-exchange' => $deadExName,
'x-message-ttl' => $ttl,
'x-dead-letter-routing-key' => $queueName
]);
$this->channel->queue_declare($queueName, false, true, false, false, false, $args);
// 创建交换机
$this->channel->exchange_declare($deadExName, AMQPExchangeType::DIRECT, false, true, false);
// 创建死信队列
$this->channel->queue_declare($deadQueueName, false, true, false, false);
// 绑定死信队列
$this->channel->queue_bind($deadQueueName, $deadExName, $queueName, false);
}
}
然后就可以写具体的业务逻辑了，这里我们要做的还是上篇内容提到的提交失败每隔几分钟重试一次，直到重试N次，结束。
const RETRY_COUNT = 10;
const RETRY_TIME = 5;
private $ttl = 1000 * 60 * self::RETRY_TIME;
private $deadExName = 'retry-exchange';// 死信exchange
private $deadQueueName = 'retry-queue';// 死信queue
private $queueName = 'ttl-retry-queue';// 延时queue
public function actionAddRetry(){
$delay = new DelayQueue();
$ttl = $this->ttl;// 1分钟后过期，放入死信队列
$deadExName = $this->deadExName;// 死信exchange
$deadQueueName = $this->deadQueueName;// 死信queue
$queueName = $this->queueName;// 延时queue
$delay->createQueue($ttl, $deadExName, $deadQueueName, $queueName);
// 模拟10个待重试的帖子
for ($i = 0; $i < 10; $i++) {
// 组装数据
$dataId = $i + 1;
$data = [
'data_id' => $dataId,
'appId' => "app_id_".$dataId,
'reviewContext' => "review_context_".$dataId,
'title' => 'title_'.$dataId,
'retry_times' => 1, // 记录重试次数
];
// 发送消息，放入延时队列，等待过期进入死信
$delay->sendMessage(json_encode($data), $queueName);
sleep(10);
}
}
public function actionConsumeRetry(){
$delay = new DelayQueue();
$deadQueueName = 'retry-queue';// 死信queue
echo date('Y-m-d H:i:s').",开始监听...". PHP_EOL;
$callback = function ($msg) {
echo $msg->body . PHP_EOL;
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
// 处理重试逻辑
$data = json_decode($msg->body,true);
// 重新提交,此处省略提交过程，假设又提交失败了
$flag = false;
if(!$flag){
if($data['retry_times'] < self::RETRY_COUNT){
$data['retry_times'] ++;
$ttl = $this->ttl;// 1分钟后过期，放入死信队列
$deadExName = $this->deadExName;// 死信exchange
$deadQueueName = $this->deadQueueName;// 死信queue
$queueName = $this->queueName;// 延时queue
$delay = new DelayQueue();
$delay->createQueue($ttl, $deadExName, $deadQueueName, $queueName);
// 重新放入延时队列
$delay->sendMessage(json_encode($data), $queueName);
}else{
echo "data_id = ".$data['data_id']."已经重试".self::RETRY_COUNT."次，结束重试".PHP_EOL;
}
}
sleep(10);
};
// 消费重试队列
$delay->consumeMessage($deadQueueName, $callback);
echo date('Y-m-d H:i:s').",结束监听". PHP_EOL;
}
}
参考：
https://segmentfault.com/a/1190000022774099/
