利用延时队列实现自动定时重试 - 基于PHP实现

测试步骤

1. 搭建代码框架
   • 创建测试项目的目录 mq
   • 从 https://github.com/php-amqplib/php-amqplib 下载AMQP库（当然也可以通过 composer 安装，这里为了简单直接自己处理了），放入 mq 目录
   • 编写 index.php，实现自动加载
2. 创建 test 目录，编写生产者和消费者
3. 跑脚本：
   • 开启生产者：

     php -f index.php retryP p

   • 开启消费者：

     php -f index.php retryC c

代码结构

├─PhpAmqpLib
│  ├─Channel
│  ├─Connection
│  ├─Exception
│  ├─Exchange
│  ├─Helper
│  │  └─Protocol
│  ├─Message
│  └─Wire
│      └─IO
├─test
│  ├─retryP.php
│  └─retryC.php
└─index.php
           

代码

index.php

<?php

function my_autoloader($cName) {
	include(__DIR__."/".$cName.".php");
}

spl_autoload_register("my_autoloader");

if (isset($argv[2])) {
	$cname = '\test\\'.$argv[1];
	if (!class_exists($cname)) {
		exit("class (".$cname.") not exists".PHP_EOL);
	}
	$c = new $cname();
	if (!method_exists($c, $argv[2])) {
		exit("method (".$argv[2].") not exists".PHP_EOL);
	}
	if (isset($argv[3])) {
		call_user_func(array($c, $argv[2]), $argv[3]);
	} else {
		call_user_func(array($c, $argv[2]));
	}
} else if (isset($argv[1])) {
	if (!function_exists($argv[1])) {
		exit("function (".$argv[1].") not exists".PHP_EOL);
	}
	$argv[1]();
} else {
	exit("please input at least one argument".PHP_EOL);
}
           

生产者

<?php

namespace test;

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

class retryP {
	private $host = 'localhost';
	private $port = 5672;
	private $user = 'guest';
	private $password = 'guest';
	
	// 可能丢失
	public function p() {
		$connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, '/', false, 'AMQPLAIN', null, 'en_US', 3.0, 120.0, null, true, 60);
		$channel = $connection->channel();

		$channel->exchange_declare('retry.retryExchange', 'direct', false, true, false, false);
		
		$channel->queue_declare('retry.normalQueue', false, true, false, false); // retryQueue 的死信队列，由消费者订阅
		$channel->queue_declare('retry.retryQueue', false, true, false, false, false, new AMQPTable(
			[
				'x-dead-letter-exchange' => 'retry.retryExchange',
				'x-dead-letter-routing-key' => 'normal',
				'x-message-ttl' => 3 * 1000
			]
		)); // retryQueue 没有消费者，超时后发到死信队列
		$channel->queue_declare('retry.failQueue', false, true, false, false); // 超过重试次数，人工处理
		
		$channel->queue_bind('retry.retryQueue', 'retry.retryExchange', 'retry');
		$channel->queue_bind('retry.normalQueue', 'retry.retryExchange', 'normal');
		$channel->queue_bind('retry.failQueue', 'retry.retryExchange', 'fail');

		// 准备消息
		$msg = new AMQPMessage(json_encode(["data"=>"something", "retryTime"=>1]));
		$channel->basic_publish($msg, 'retry.retryExchange', 'normal');
		echo "basic_publish success";

		$channel->close();
		$connection->close();
	}
}
           

消费者

<?php

namespace test;

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

class retryC {
	private $host = 'localhost';
	private $port = 5672;
	private $user = 'guest';
	private $password = 'guest';
	
	private $connection;
	private $channel;
	
	public function __construct() {
		$this->connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, '/', false, 'AMQPLAIN', null, 'en_US', 3.0, 120.0, null, true, 60);
		$this->channel = $this->connection->channel();
	}
	
	// 正常队列
	public function c() {		
		//闭包回调函数
		$callback = function ($msg) {
			echo $msg->body;
			echo PHP_EOL;
			$data = json_decode($msg->body, true);
			// 超过重试次数，放入死信队列
			if ($data["retryTime"] > 3) {
				$msg->delivery_info['channel']->basic_publish($msg, 'retry.retryExchange', 'fail');
				echo "retryTime > 3, fail" . PHP_EOL;
			}
			// 消费消息，进行业务处理
			$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
			// 业务处理失败，向 retryQueue 发消息
			$msgNew = new AMQPMessage(json_encode(["data"=>$data["data"], "retryTime"=>$data["retryTime"]+1]));
			$msg->delivery_info['channel']->basic_publish($msgNew, 'retry.retryExchange', 'retry');
		};
		$this->channel->basic_qos(null, 1, null);
		$this->channel->basic_consume('retry.normalQueue', '', false, false, false, false, $callback); // 需要手动确认

		while (count($this->channel->callbacks)) {
			$this->channel->wait();
		}
		$this->channel->close();
		$this->connection->close();
	}
}