背景
Aliware
在企业的商业活动中,订单是指交易双方的产品或服务交易意向。交易下单负责创建这个交易双方的产品或服务交易意向,有了这个意向后,买方可以付款,卖方可以发货。
买家超时未付款:比如超过15分钟没有支付,订单自动取消。
商家超时未发货:比如商家超过1个月没发货,订单自动取消。
买家超时未收货:比如商家发货后,买家没有在14天内点击确认收货,则系统默认自动收货。
JDK 自带的延时队列
Aliware
JDK中提供了一种延迟队列数据结构DelayQueue,其本质是封装了PriorityQueue,可以把元素进行排序。
把订单插入DelayQueue中,以超时时间作为排序条件,将订单按照超时时间从小到大排序。
起一个线程不停轮询队列的头部,如果订单的超时时间到了,就出队进行超时处理,并更新订单状态到数据库中。
优点:简单,不需要借助其他第三方组件,成本低。
缺点:
所有超时处理订单都要加入到DelayQueue中,占用内存大。
没法做到分布式处理,只能在集群中选一台leader专门处理,效率低。
RabbitMQ的延时消息
Aliware
RabbitMQ的延时消息主要有两个解决方案:
RabbitMQ Delayed Message Plugin
TTL:即消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL,如果对队列设置,则队列中所有的消息都具有相同的过期时间。超过了这个时间,我们认为这个消息就死了,称之为死信。
死信Exchange(DLX):一个消息在满足以下条件会进入死信交换机
一个消息被Consumer拒收了,并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里,被其他消费者使用。
TTL到期的消息。
定义一个BizQueue,用来接收死信消息,并进行业务消费。
定义一个死信交换机(DLXExchange),绑定BizQueue,接收延时队列的消息,并转发给BizQueue。
定义一组延时队列DelayQueue_xx,分别配置不同的TTL,用来处理固定延时5s、10s、30s等延时等级,并绑定到DLXExchange。
优点:可以支持海量延时消息,支持分布式处理。
缺点:
不灵活,只能支持固定延时等级。
RocketMQ的定时消息
Aliware
MessageBuilder messageBuilder = null;
Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000; //延迟10分钟
Message message = messageBuilder.setTopic("topic")
//设置消息索引键,可根据关键字精确查找某条消息。
.setKeys("messageKey")
//设置消息Tag,用于消费端根据指定Tag过滤消息。
.setTag("messageTag")
//设置延时时间
.setDeliveryTimestamp(deliverTimeStamp)
//消息体
.setBody("messageBody".getBytes())
.build();
SendReceipt sendReceipt = producer.send(message);
System.out.println(sendReceipt.getMessageId());
优点
精度高,支持任意时刻。
使用门槛低,和使用普通消息一样。
缺点
使用限制:定时时长最大值24小时。
成本高:每个订单需要新增一个定时消息,且不会马上消费,给MQ带来很大的存储成本。
Redis的过期监听
Aliware
Redis支持过期监听,也能达到和RocketMQ定时消息一样的能力,具体步骤如下:
redis配置文件开启"notify-keyspace-events Ex"
public class RedisListenerConfig {
RedisMessageListenerContainer container(RedisConnectionFactory factory){
RedisMessageListenerContainer container=new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
return container;
}
}
public class RedisKeyExpirationListerner extends KeyExpirationEventMessageListener {
public RedisKeyExpirationListerner(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
public void onMessage(Message message, byte[] pattern) {
String keyExpira = message.toString();
System.out.println("监听到key:" + expiredKey + "已过期");
}
}
typedef struct redisDb {
dict *dict; /* 维护所有key-value键值对 */
dict *expires; /* 过期字典,维护设置失效时间的键 */
....
} redisDb;
过期字典本质上是一个链表,每个节点的数据结构结构如下:
key是一个指针,指向某个键对象。
value是一个long long类型的整数,保存了key的过期时间。
定期删除:每隔一段时间(默认100ms)就随机抽取一些设置了过期时间的key,检查其是否过期,如果有过期就删除。之所以这么做,是为了通过限制删除操作的执行时长和频率来减少对cpu的影响。不然每隔100ms就要遍历所有设置过期时间的key,会导致cpu负载太大。
惰性删除:不主动删除过期的key,每次从数据库访问key时,都检测key是否过期,如果过期则删除该key。惰性删除有一个问题,如果这个key已经过期了,但是一直没有被访问,就会一直保存在数据库中。
定时任务分布式批处理
Aliware
稳定性强:基于通知的方案(比如MQ和Redis),比较担心在各种极端情况下导致通知的事件丢了。使用定时任务跑批,只需要保证业务幂等即可,如果这个批次有些订单没有捞出来,或者处理订单的时候应用重启了,下一个批次还是可以捞出来处理,稳定性非常高。
效率高:基于MQ的方案,需要一个订单一个定时消息,consumer处理定时消息的时候也需要一个订单一个订单更新,对数据库tps很高。使用定时任务跑批方案,一次捞出一批订单,处理完了,可以批量更新订单状态,减少数据库的tps。在海量订单处理场景下,批量处理效率最高。
可运维:基于数据库存储,可以很方便的对订单进行修改、暂停、取消等操作,所见即所得。如果业务跑失败了,还可以直接通过sql修改数据库来进行批量运维。
成本低:相对于其他解决方案要借助第三方存储组件,复用数据库的成本大大降低。
通过实现map函数,通过代码自行构造分片,SchedulerX会将分片平均分给超时中心的不同节点分布式执行。
通过实现reduce函数,可以做聚合,可以判断这次跑批有哪些分片跑失败了,从而通知下游处理。
免运维、成本低:不需要自建任务调度系统,由云上托管。
可观测:提供任务执行的历史记录、查看堆栈、日志服务、链路追踪等能力。
高可用:支持同城双活容灾,支持多种渠道的监控报警。
混部:可以托管阿里云的机器,也可以托管非阿里云的机器。
总结
Aliware
如果对于超时精度比较高,超时时间在24小时内,且不会有峰值压力的场景,推荐使用RocketMQ的定时消息解决方案。
[1] https://developer.aliyun.com/article/994932
[2] https://redis.io/docs/manual/keyspace-notifications/
[3] https://www.aliyun.com/aliware/schedulerx
作者:阿里巴巴中间件
本文为 @ 阿里巴巴中间件 创作并授权 21CTO 发布,未经许可,请勿转载。
内容授权事宜请您联系 webmaster@21cto.com或关注 21CTO 公众号。
该文观点仅代表作者本人,21CTO 平台仅提供信息存储空间服务。