今天和大家分享一个简单的rabbit实现延迟任务的方法,rabbit实现延迟队列有两种方式,一种是队列或者消息的TTL(Time To Live),另一种是rabbit的rabbitmq-delayed-message-exchange插件,今天我和大家分享下TTL的使用方法。
rabbit有Per-Queue Message TTL和Per-Message TTL两种设置超时的方式,分别指针对消息和队列的,给消息添加过期时间相对比较灵活,这样不用每一种过期时间都去建立一个队列去监听,给消息设置过期时间方法
$msg = new AMQPMessage('hello expiration!');
$msg->set("expiration","5000"); //关键一点:超时时间必须设置成字符串,否则不会生效!单位是ms
好了这会儿大家会想超时之后的处理在哪写呢,总有个超时的回调或者什么吧,rabbit可以给队列设置一个x-dead-letter-exchange,Dead letter routing key,意思是消息超时后的转发队列。
$channel->queue_declare("waitSendQueue",false,false,false,false,false,new AMQPTable(array
("x-dead-letter-exchange"=>"expireExchange")));
waitSendQueue上的消息如果超时了会转发给expireExchange,那我只要去监听expireExchange上的消息,拿到消息去处理业务,就完成了我们的延迟任务,那么接下来上代码.。
客户端:
/**
* Created by PhpStorm.
* User: qyc
* Date: 2017/8/14
* Time: 下午2:18
*/
require_once dirname(__DIR__) . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
$con = new AMQPStreamConnection('localhost', 5672, "guest", "guest");
$channel = $con->channel();
//定义等待exchange
$channel->exchange_declare('waitSendExchange', 'fanout', false, false, false);
//定义过期exchange
$channel->exchange_declare('expireExchange', 'fanout', false, false, false);
//定义过期queue
$channel->queue_declare("expireQueue",false,false,false,false,false);
//定义等待queue
$channel->queue_declare("waitSendQueue",false,false,false,false,false,new AMQPTable(array
("x-dead-letter-exchange"=>"expireExchange")));
$channel->queue_bind("waitSendQueue","waitSendExchange");
$channel->queue_bind("expireQueue","expireExchange");
$msg = new AMQPMessage('hello expiration!');
/*
* 设置超时时间
*/
$msg->set("expiration","5000");
/**
* 向等待exchage发布消息
*/
$channel->basic_publish($msg, 'waitSendExchange');
echo 'send1:' . date('Y-m-d H:i:s') . "\n";
$channel->close();
$con->close();
服务端
/**
* Created by PhpStorm.
* User: qyc
* Date: 2017/8/14
* Time: 下午2:30
*/
require_once dirname(__DIR__) . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
$con = new AMQPStreamConnection("localhost",5672,"guest","guest");
$channel = $con->channel();
$channel = $con->channel();
//定义等待exchange
$channel->exchange_declare('waitSendExchange', 'fanout', false, false, false);
//定义过期exchange
$channel->exchange_declare('expireExchange', 'fanout', false, false, false);
//定义过期queue
$channel->queue_declare("expireQueue",false,false,false,false,false);
//定义等待queue
$channel->queue_declare("waitSendQueue",false,false,false,false,false,new AMQPTable(array
("x-dead-letter-exchange"=>"expireExchange")));
$channel->queue_bind("waitSendQueue","waitSendExchange");
$channel->queue_bind("expireQueue","expireExchange");
$callback = function ($msg){
echo "\n".' [x] ', $msg->body, "\n";
echo "receive2:".date("Y-m-d H:i:s")."\n";
};
/**
* 订阅超时queue
*/
$channel->basic_consume("expireQueue","",false,true,false,false,$callback);
while (count($channel->callbacks)){
$channel->wait();
}
$channel->close();
$connection->close();
结果:
发送:
接受: