基于 rabbitmq 实现延迟队列
2017-02-09 17:17
831 查看
roc • 0 回帖 • 10 浏览
• 2 周前
platform
< >
##
延迟队列能做什么?
###
开发过程中通常会碰到这样的需求:
淘宝订单业务:下单后 30min 之内没有付款,就自动取消订单。
饿了吗订餐通知:下单成功后 60s 之后给用户发送短信通知。
关闭空闲连接:服务器中有很多客户端的连接,空闲一段时间之后需要关闭之。
缓存:缓存中的对象,超过了空闲时间,从缓存中移出。
任务超时处理:在网络协议滑动窗口请求应答式交互时,处理超时未响应的请求。
失败重试机制:业务操作失败后,间隔一定的时间进行失败重试。
这类业务的特点就是:需要延迟工作,需要进行失败重试。一种比较笨的方式是使用一个后台线程,遍历所有对象,挨个检查。这种方法简单好用,但是对象数量过多时,可能存在性能问题,检查间隔时间不好设置,间隔时间过大,影响精确度,过小则存在效率问题,而且做不到按超时的时间顺序处理。
本地延迟队列
DelayQueue
###DelayQueue 是一个无界的 BlockingQueue,用于放置实现了 Delayed 接口的对象,其中的对象只能在其到期时才能从队列中取走。这种队列是有序的,即队头对象的延迟到期时间最长。Delayed 扩展了 Comparable 接口,比较的基准为延时的时间值,Delayed 接口的实现类 getDelay 的返回值应为固定值(final)。DelayQueue 内部是使用 PriorityQueue 实现的。
DelayQueue = BlockingQueue + PriorityQueue + Delayed
DelayQueue 的关键元素 BlockingQueue、PriorityQueue、Delayed。可以这么说,DelayQueue 是一个使用优先队列(PriorityQueue)实现的 BlockingQueue,优先队列的比较基准值是时间。注意:不能将 null 元素放置到这种队列中。
但是我们知道,利用 DelayQueue 实现的是一个单机的、JVM 内存中的延迟队列,并没有集群的支持,而且无法满足在对业务系统泵机的时、消息消费异常的时候做相应的逻辑处理。
基于分布式消息队列
RabbitMQ 实现延迟队列
###RabbitMQ 本身没有直接支持延迟队列功能,但是可以通过以下特性模拟出延迟队列的功能:
1>
Per-Queue Message TTL RabbitMQ 可以对消息和队列设置 TTL(过期时间)。
###RabbitMQ 针对队列中的消息过期时间(Time To Live, TTL)有两种方法可以设置。第一种方法是通过队列属性设置,队列中所有消息都有相同的过期时间。第二种方法是对消息进行单独设置,每条消息 TTL 可以不同。如果上述两种方法同时使用,则消息的过期时间以两者之间 TTL 较小的那个数值为准。消息在队列的生存时间一旦超过设置的 TTL 值,就成为 dead message, 消费者将无法再收到该消息。
2>
Dead Letter Exchanges 死信消息
###利用 DLX,当消息在一个队列中变成死信后,它能被重新 publish 到另一个 Exchange,这个 Exchange 就是 DLX。消息变成死信有以下几种情况:
消息被拒绝(basic.reject or basic.nack)并且 requeue=false
消息 TTL 过期
队列达到最大长度
DLX 也是一下正常的 Exchange 同一般的 Exchange 没有区别,它能在任何的队列上被指定,实际上就是设置某个队列的属性,当这个队列中有死信时,RabbitMQ 就会自动的将这个消息重新发布到设置的 Exchange 中去,进而被路由到另一个队列,publish 可以监听这个队列中消息做相应的处理,这个特性可以弥补 RabbitMQ 3.0.0 以前支持的 immediate 参数中的向 publish
确认的功能。
结合以上两个特性,就可以模拟出延迟消息的功能.
流程图
源代码
package hbec.app.stock.rabbitmq.utils;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
/**
* @Description 基于 RabbitMQ 实现的分布式延迟重试队列
*
*
*
delayExchangeName : 交换器名称
*
delayQueueName : 延迟队列名称
*
delayRoutingKeyName : 路由器名称
*
perDelayQueueMessageTTL : 延迟队列中 message 的默认 ttl
*
* 通过{@link RabbitMQDelayQueue#put(byte[], long, TimeUnit)}首次进入延迟队列的消息,
* 其 ttl = min(message ttl, per queue message ttl),
* 消息被 Reject/nack 之后变成死信消息,其自带 message ttl 失效,
* 以后将按照{@link #perDelayQueueMessageTTL}指定的延迟时间投递给经由{@link RabbitMQDelayQueue#consumerRegister}注册的消费者,直到消息被 Ack.
*
* @author roc roc.fly@qq.com
* @date Dec 9, 2016 3:29:39 PM
*/
public class RabbitMQDelayQueue {
private static Logger LOGGER = LoggerFactory.getLogger(RabbitMQDelayQueue.class); private static final String POSTFIX_TASK = "_task"; // direct类型 交换器 public static final String EXCHANGE_TYPE_DIRECT = "direct"; private Connection connection; //注册消费者 private ConsumerRegister consumerRegister; //任务队列配置 private String taskExchangeName; private String taskQueueName; private String taskRoutingKeyName; //延迟队列配置 private String delayExchangeName; private String delayQueueName; private String delayRoutingKeyName; //延迟队列中的消息ttl private long perDelayQueueMessageTTL; public RabbitMQDelayQueue(Connection connection, ConsumerRegister consumerRegister, String delayExchangeName, String delayQueueName, String delayRoutingKeyName, long perDelayQueueMessageTTL) throws IOException { this.connection = connection; this.consumerRegister = consumerRegister; this.delayExchangeName = delayExchangeName; this.delayQueueName = delayQueueName; this.delayRoutingKeyName = delayRoutingKeyName; this.perDelayQueueMessageTTL = perDelayQueueMessageTTL; this.taskExchangeName = delayExchangeName + POSTFIX_TASK; this.taskQueueName = delayQueueName + POSTFIX_TASK; this.taskRoutingKeyName = delayRoutingKeyName + POSTFIX_TASK; init(); registerConsumer(); } /** * * @Description 注册消费者 * @author roc roc.fly@qq.com * @date Dec 29, 2016 1:36:25 PM */ public interface ConsumerRegister { public Consumer register(Channel channel) throws IOException; } /** * 注册带有ttl的queue和对应的任务队列 * * @throws IOException * @author roc */ private void init() throws IOException { Channel channel = connection.createChannel(); channel.exchangeDeclare(taskExchangeName, EXCHANGE_TYPE_DIRECT, true); channel.exchangeDeclare(delayExchangeName, EXCHANGE_TYPE_DIRECT, true); // 任务队列 B HashMap<String, Object> argumentsTask = Maps.newHashMap(); argumentsTask.put("x-dead-letter-exchange", delayExchangeName); argumentsTask.put("x-dead-letter-routing-key", delayRoutingKeyName); channel.queueDeclare(taskQueueName, true, false, false, argumentsTask); channel.queueBind(taskQueueName, taskExchangeName, taskRoutingKeyName); // 延迟队列 A HashMap<String, Object> argumentsDelay = Maps.newHashMap(); argumentsDelay.put("x-dead-letter-exchange", taskExchangeName); argumentsDelay.put("x-dead-letter-routing-key", taskRoutingKeyName); argumentsDelay.put("x-message-ttl", perDelayQueueMessageTTL); channel.queueDeclare(delayQueueName, true, false, false, argumentsDelay); channel.queueBind(delayQueueName, delayExchangeName, delayRoutingKeyName); channel.close(); } /** * 注册消费者 * @throws IOException * @author roc */ private void registerConsumer() throws IOException { LOGGER.info("register consumer ->{}", this); Channel channel = connection.createChannel(); Consumer consumer = consumerRegister.register(channel); channel.basicConsume(taskQueueName, false, consumer); LOGGER.info("register consumer ->{} success", this); } /** * 消息入队 * * @param body 消息内容 * @param timeout 超时时间 * @param unit 超时时间单位 * @throws IOException * @author roc */ public void put(byte[] body, long timeout, TimeUnit unit) throws IOException { Preconditions.checkNotNull(body); Preconditions.checkArgument(timeout >= 0); Preconditions.checkNotNull(unit); LOGGER.info("put element to delay queue ->{}", body.hashCode()); Channel channel = null; try { channel = connection.createChannel(); // deliveryMode=2 标识任务的持久性 long millis = unit.toMillis(timeout); AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().expiration(String.valueOf(millis)).deliveryMode(2).build(); channel.basicPublish(delayExchangeName, delayRoutingKeyName, properties, body); LOGGER.info("put element to delay queue success"); } finally { if (null != channel) channel.close(); } } public static class Builder { private Connection connection; private ConsumerRegister consumerRegister; private String delayExchangeName; private String delayQueueName; private String delayRoutingKeyName; private long perDelayQueueMessageTTL; public Builder setConnection(Connection connection) { this.connection = connection; return this; } public Builder setDelayExchangeName(String delayExchangeName) { this.delayExchangeName = delayExchangeName; return this; } public Builder setDelayQueueName(String delayQueueName) { this.delayQueueName = delayQueueName; return this; } public Builder setDelayRoutingKeyName(String delayRoutingKeyName) { this.delayRoutingKeyName = delayRoutingKeyName; return this; } public Builder setConsumerRegister(ConsumerRegister consumerRegister) { this.consumerRegister = consumerRegister; return this; } public Builder setPerDelayQueueMessageTTL(long timeout, TimeUnit unit) { this.perDelayQueueMessageTTL = unit.toMillis(timeout);; return this; } public RabbitMQDelayQueue build() throws IOException { Preconditions.checkNotNull(connection); Preconditions.checkNotNull(delayExchangeName); Preconditions.checkNotNull(delayQueueName); Preconditions.checkNotNull(delayRoutingKeyName); Preconditions.checkNotNull(consumerRegister); return new RabbitMQDelayQueue(connection, consumerRegister, delayExchangeName, delayQueueName, delayRoutingKeyName, perDelayQueueMessageTTL); } }
}
测试代码
package hbec.app.stock.rabbitmq.utils;
import hbec.app.stock.rabbitmq.utils.RabbitMQDelayQueue.ConsumerRegister;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 测试 demo
*
*/
public class RabbitMQDelayQueueTest {
public static void main(String[] args) throws IOException { delayQueue(); } public static void delayQueue() throws IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("guest"); factory.setPassword("guest"); Address address = new Address("10.0.30.67", 56720); Connection connection = factory.newConnection(new Address[] { address }); RabbitMQDelayQueue delayQueue = new RabbitMQDelayQueue.Builder().setConnection(connection).setPerDelayQueueMessageTTL(15, TimeUnit.SECONDS).setDelayExchangeName("delay_exchange_roc").setDelayQueueName("delay_queue_roc").setDelayRoutingKeyName("delay_routing_key_roc").setConsumerRegister(new ConsumerRegister() { @Override public Consumer register(Channel channel) throws IOException { return new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); String routingKey = envelope.getRoutingKey(); // TODO do something String content = new String(body, Charset.forName("utf-8")); System.out.println("receive message --- > " + content); Map<String, Object> headers = properties.getHeaders(); if (headers != null) { List<Map<String, Object>> xDeath = (List<Map<String, Object>>) headers.get("x-death"); System.out.println("xDeath--- > " + xDeath); if (xDeath != null && !xDeath.isEmpty()) { Map<String, Object> entrys = xDeath.get(0); } } // 消息拒收 // if(do something) 消息重新入队 getChannel().basicReject(deliveryTag, false); // else 消息应答 // getChannel().basicAck(deliveryTag, false); } }; } }).build(); delayQueue.put("{\"name\" : \"i am roc!!\"}\"".getBytes("UTF-8"), 3, TimeUnit.SECONDS); }
}
码字不容易啊,觉得有用的请我喝咖啡~~
相关文章推荐
- 基于 rabbitmq 实现延迟队列
- 如何基于RabbitMQ实现优先级队列
- rabbitmq 实现延迟队列(ttl+DLX)
- rabbitmq 延迟队列的实现(PHP)http://blog.yuhai.win
- RabbitMQ 延迟队列实现订单支付结果异步阶梯性通知
- Spring MVC整合RabbitMQ实现延迟队列
- 如何基于RabbitMQ实现优先级队列
- C#实现rabbitmq 延迟队列功能
- C#实现rabbitmq 延迟队列功能
- C#实现rabbitmq 延迟队列功能实例代码
- 基于Redis实现延迟队列
- 如何基于RabbitMQ实现优先级队列
- RabbitMQ如何实现延迟队列?
- 基于PHP使用rabbitmq实现消息队列
- 跨数据库分布式实时事务 - 基于RabbitMQ实时消息队列服务实现
- 如何基于RabbitMQ实现优先级队列
- 如何基于RabbitMQ实现优先级队列
- rabbitmq 实现延迟队列的两种方式
- Spring集成RabbitMQ并实现延迟队列
- Spring Boot与RabbitMQ结合实现延迟队列的示例