rabbitmq 延时队列 插件方式实现 每条消息都延时自己时间

时间：2024-04-23 08:36:36

上篇文章的延时是加到队列上的 通过死信过时推送 ，缺点就是不能每条消息定义自己的过时时间而且每次有新的过时时间，要新建一个交换机和队列

https://www.cnblogs.com/brady-wang/p/13335104.html

rabbitmq还有种方式 要安装一个插件  rabbitmq-delayed-message-exchange

参考  https://www.cnblogs.com/brady-wang/p/13335243.html

实现是安装插件后交换机会多出一种 不过这种插件要安装 好像mq版本至少3.7

rabbitmq 延时队列 插件方式实现 每条消息都延时自己时间

最终生产者生产时候 头部加上延时时间，那么他会存储在交换机里面，到时了才投递到对应队列

$exchange->setType('x-delayed-message'); //x-delayed-message类型

$exchange->publish($message, $params['routeKey'], AMQP_NOPARAM, ['headers'=>['x-delay'=> 1000*$i]]);
[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 14:05:25

i=3600，延迟3600秒

[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 14:05:28

i=3600，延迟3600秒

[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 14:05:30

i=3600，延迟3600秒

[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 18:52:21

i=3600，延迟3600秒

[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 18:52:48

i=3600，延迟3600秒

^[[A

[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 18:52:51

i=3600，延迟3600秒

^[[A[root@localhost mq]# php delay_publish.php

发送时间：2020-07-18 18:52:54

i=3600，延迟3600秒

　　

[root@localhost mq]# php delay_comsumer.php

接收时间：2020-07-18 13:58:11

接收内容：{"order_id":1595051842,"i":5,"date":"2020-07-18 13:57:22"}

接收时间：2020-07-18 13:58:11

接收内容：{"order_id":1595051844,"i":4,"date":"2020-07-18 13:57:24"}

接收时间：2020-07-18 13:58:11

接收内容：{"order_id":1595051846,"i":3,"date":"2020-07-18 13:57:26"}

接收时间：2020-07-18 13:58:11

接收内容：{"order_id":1595051848,"i":2,"date":"2020-07-18 13:57:28"}

接收时间：2020-07-18 13:58:11

接收内容：{"order_id":1595051850,"i":1,"date":"2020-07-18 13:57:30"}

rabbitmq 延时队列 插件方式实现 每条消息都延时自己时间

可以看到里面有4条消息延时了

代码 如下

生产者 delay_publish.php

<?php

//header('Content-Type:text/html;charset=utf-8;');

$params = array(

	'exchangeName' => 'delayed_exchange_test',

	'queueName' => 'delayed_queue_test',

	'routeKey' => 'delayed_route_test',

);

$connectConfig = array(

	'host' => '127.0.0.1',

	'port' => 5672,

	'login' => 'admin',

	'password' => 'password',

	'vhost' => '/'

);

//var_dump(extension_loaded('amqp')); 判断是否加载amqp扩展

//exit();

try {

	$conn = new AMQPConnection($connectConfig);

	$conn->connect();

	if (!$conn->isConnected()) {

		//die('Conexiune esuata');

		//TODO 记录日志

		echo 'rabbit-mq 连接错误:', json_encode($connectConfig);

		exit();

	}

	$channel = new AMQPChannel($conn);

	if (!$channel->isConnected()) {

		// die('Connection through channel failed');

		//TODO 记录日志

		echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);

		exit();

	}

	$exchange = new AMQPExchange($channel);

	$exchange->setName($params['exchangeName']);

	$exchange->setType('x-delayed-message'); //x-delayed-message类型

	/*RabbitMQ常用的Exchange Type有三种：fanout、direct、topic。

      fanout:把所有发送到该Exchange的消息投递到所有与它绑定的队列中。

      direct:把消息投递到那些binding key与routing key完全匹配的队列中。

      topic:将消息路由到binding key与routing key模式匹配的队列中。*/

	$exchange->setArgument('x-delayed-type','direct');

	$exchange->declareExchange();

	//$channel->startTransaction();

	//RabbitMQ不容许声明2个相同名称、配置不同的Queue,否则报错

	$queue = new AMQPQueue($channel);

	$queue->setName($params['queueName']);

	$queue->setFlags(AMQP_DURABLE);

	$queue->declareQueue();

	//绑定队列和交换机

	$queue->bind($params['exchangeName'], $params['routeKey']);

	//$channel->commitTransaction();

} catch(Exception $e) {

}

//for($i=5;$i>0;$i--){

$i = 3600;

	//生成消息

	echo '发送时间：'.date("Y-m-d H:i:s", time()).PHP_EOL;

	echo 'i='.$i.'，延迟'.$i.'秒'.PHP_EOL;

	$message = json_encode(['order_id'=>time(),'i'=>$i,'date'=>date("Y-m-d H:i:s")]);

	$exchange->publish($message, $params['routeKey'], AMQP_NOPARAM, ['headers'=>['x-delay'=> 1000*$i]]);

	sleep(2);

//}

$conn->disconnect();

　　

消费者 delay_consumer.php

<?php

//header('Content-Type:text/html;charset=utf8;');

$params = array(

	'exchangeName' => 'delayed_exchange_test',

	'queueName' => 'delayed_queue_test',

	'routeKey' => 'delayed_route_test',

);

$connectConfig = array(

	'host' => '127.0.0.1',

	'port' => 5672,

	'login' => 'admin',

	'password' => 'password',

	'vhost' => '/'

);

//var_dump(extension_loaded('amqp'));

//exit();

try {

	$conn = new AMQPConnection($connectConfig);

	$conn->connect();

	if (!$conn->isConnected()) {

		//die('Conexiune esuata');

		//TODO 记录日志

		echo 'rabbit-mq 连接错误:', json_encode($connectConfig);

		exit();

	}

	$channel = new AMQPChannel($conn);

	if (!$channel->isConnected()) {

		// die('Connection through channel failed');

		//TODO 记录日志

		echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);

		exit();

	}

	$exchange = new AMQPExchange($channel);

	//$exchange->setFlags(AMQP_DURABLE);//声明一个已存在的交换器的，如果不存在将抛出异常，这个一般用在consume端

	$exchange->setName($params['exchangeName']);

	$exchange->setType('x-delayed-message'); //x-delayed-message类型

	/*RabbitMQ常用的Exchange Type有三种：fanout、direct、topic。

      fanout:把所有发送到该Exchange的消息投递到所有与它绑定的队列中。

      direct:把消息投递到那些binding key与routing key完全匹配的队列中。

      topic:将消息路由到binding key与routing key模式匹配的队列中。*/

	$exchange->setArgument('x-delayed-type','direct');

	$exchange->declareExchange();

	//$channel->startTransaction();

	$queue = new AMQPQueue($channel);

	$queue->setName($params['queueName']);

	$queue->setFlags(AMQP_DURABLE);

	$queue->declareQueue();

	//绑定

	$queue->bind($params['exchangeName'], $params['routeKey']);

} catch(Exception $e) {

	echo $e->getMessage();

	exit();

}

function callback(AMQPEnvelope $message) {

	global $queue;

	if ($message) {

		$body = $message->getBody();

		echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;

		echo '接收内容：'.$body . PHP_EOL;

		//为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息

		$queue->ack($message->getDeliveryTag());

	} else {

		echo 'no message' . PHP_EOL;

	}

}

//$queue->consume('callback');  第一种消费方式,但是会阻塞,程序一直会卡在此处

//第二种消费方式,非阻塞

/*$start = time();

while(true)

{

    $message = $queue->get();

    if(!empty($message))

    {

        echo $message->getBody();

        $queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费

        $end = time();

        echo '<br>' . ($end - $start);

        exit();

    }

    else

    {

        //echo 'message not found' . PHP_EOL;

    }

}*/

//注意：这里需要注意的是这个方法：$queue->consume，queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；后者则是非阻塞的，取消息时有则取，无则返回false。

//就是说用了consume之后，会同步阻塞，该程序常驻内存，不能用nginx，apache调用。

$action = '2';

if($action == '1'){

	$queue->consume('callback');  //第一种消费方式,但是会阻塞,程序一直会卡在此处

}else{

	//第二种消费方式,非阻塞

	$start = time();

	while(true)

	{

		$message = $queue->get();

		if(!empty($message))

		{

			echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;

			echo '接收内容：'.$message->getBody().PHP_EOL;

			$queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费

			$end = time();

			//echo '运行时间：'.($end - $start).'秒'.PHP_EOL;

			//exit();

		}

		else

		{

			//echo 'message not found' . PHP_EOL;

		}

	}

}

　　

相关文章