天水网站建设公司,网站开发欠款,wordpress 帝国 seo,装饰设计网站概述
业务中经常会遇到一些延迟执行的需求#xff1b;通常想到的都是rabbitmq或者rocketmq的延迟消息#xff1b; 但是系统中不一定集成了mq#xff0c;但为了控制分布式下的并发#xff0c;一般redis都是有集成的#xff1b; redis的key过期监听那个时间不准确#xff…概述
业务中经常会遇到一些延迟执行的需求通常想到的都是rabbitmq或者rocketmq的延迟消息 但是系统中不一定集成了mq但为了控制分布式下的并发一般redis都是有集成的 redis的key过期监听那个时间不准确在集群环境下节点挂了也容易丢失
那么用redisson的延迟队列正好可以用来解决轻量级的延时消息; 简单的来说就是消费者生产了一个消息任务塞到ZSet里(用当前时间戳延迟时间作为分数)等时间到了就会放到任务List中然后消费者真正去执行任务都是从任务List中获取任务
redisson中的消费者并不是一直轮询获取任务而是有具体时间的延迟任务时间到了去任务队列中获取任务
redisson延时任务机制简述
生产者先将任务push到delay_queue_timeout等待队列中延迟时间到了消费者会把任务从timeout队列挪到SANYOU任务队列中消费者实际获取任务的队列然后消费者就能拿到最终要执行的任务了;
这里具体要说的就是客户端通知和获取机制 消费者在启动时通常都会去get一下队列,达到订阅队列的目的
RBlockingQueueString blockingQueue redissonClient.getBlockingQueue(SANYOU);
RDelayedQueueString delayQueue redissonClient.getDelayedQueue(blockingQueue);这样做的目的 消费者订阅队列从delay_queue_timeout等待延迟队列中将已经到达时间的任务挪到真正的任务List队列中然后再将delay_queue_timeout队列中第一个也就是第一个要执行的的任务的时间拿到用这个时间开启一个延迟任务时间到了之后会发布一个消息到时间通知channel中然后客户端监听到这个channel中的消息后会再次重复上述步骤让delay_queue_timeout中的任务可以都放到真正的任务List队列中
这样有一个好处就是不用一直while扫描等待客户端的延迟任务时间和delay_queue_timeout中的延迟时间是一样的可以精准利用cpu理论上是没有延迟的但是实际消息数量大量增加消费者消费比较慢还是会造成延迟任务消费延迟;
另外由于客户端都是用lua脚本去redis的同一个List队列中获取任务lua脚本在redis中都是原子任务而且redis真正的操作是单线程的所以不会存在任务广播情况并发获取时一个任务不会被多个消费者同时拿到
捞一张图片
代码Demo import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;Slf4j
Component
public class RedissonDelayQueueConfig implements InitializingBean {Resourceprivate RedissonClient redissonClient;//延时队列mapprivate final MapString, RDelayedQueueDelayMessageDTO delayQueueMap new ConcurrentHashMap(16);/*** 消费者初始化所有队列,订阅对应的队列并开启第一个过期任务的过期时间对应的延迟任务*/PostConstructpublic void reScheduleDelayedTasks() {DelayQueueEnum[] queueEnums DelayQueueEnum.values();for (DelayQueueEnum queueEnum : queueEnums) {RBlockingDequeObject blockingDeque redissonClient.getBlockingDeque(queueEnum.getCode());RDelayedQueueObject delayedQueue redissonClient.getDelayedQueue(blockingDeque);}}Overridepublic void afterPropertiesSet() {// 有新的延迟队列在这里添加队列消费类需要继承DelayQueueConsumerDelayQueueEnum[] queueEnums DelayQueueEnum.values();for (DelayQueueEnum queueEnum : queueEnums) {DelayQueueConsumer delayQueueConsumer SpringUtil.getBean(queueEnum.getBeanName());if (delayQueueConsumer null) {throw new ServiceException(queueName queueEnum.getBeanName() ,delayQueueConsumernull,请检查配置...);}// Redisson的延时队列是对另一个队列的再包装使用时要先将延时消息添加到延时队列中当延时队列中的消息达到设定的延时时间后// 该延时消息才会进行进入到被包装队列中因此我们只需要对被包装队列进行监听即可。RBlockingQueueDelayMessageDTO rBlockingQueue redissonClient.getBlockingDeque(queueEnum.getCode());//消费者初始化队列RDelayedQueueDelayMessageDTO rDelayedQueue redissonClient.getDelayedQueue(rBlockingQueue);//set到map中方便获取delayQueueMap.put(queueEnum.getCode(), rDelayedQueue);// 订阅新元素的到来调用的是takeAsync()异步执行rBlockingQueue.subscribeOnElements(delayQueueConsumer::execute);}}public RedissonClient getRedissonClient() {return redissonClient;}public MapString, RDelayedQueueDelayMessageDTO getDelayQueueMap() {return delayQueueMap;}
}import cn.hutool.core.date.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;import javax.annotation.Resource;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;Slf4j
Component
public class DelayQueueUtil {private static RedissonDelayQueueConfig redissonDelayQueueConfig;Resourcepublic void setRedissonDelayQueueConfig(RedissonDelayQueueConfig redissonDelayQueueConfig) {DelayQueueUtil.redissonDelayQueueConfig redissonDelayQueueConfig;}private static MapString, RDelayedQueueDelayMessageDTO getDelayQueueMap() {if(null redissonDelayQueueConfig) return Collections.emptyMap();return redissonDelayQueueConfig.getDelayQueueMap();}private static RedissonClient getRedissonClient() {if(null redissonDelayQueueConfig) return null;return redissonDelayQueueConfig.getRedissonClient();}/*** 添加延迟消息*/public static void addDelayMessage(DelayMessageDTO delayMessage) {log.info(delayMessage{}, delayMessage);Assert.isTrue(getDelayQueueMap().containsKey(delayMessage.getQueueName()), 队列不存在);delayMessage.setCreateTime(DateUtil.now());if(null delayMessage.getTimeUnit()){delayMessage.setTimeUnit(TimeUnit.SECONDS);}RDelayedQueueDelayMessageDTO rDelayedQueue getDelayQueueMap().get(delayMessage.getQueueName());//移除相同的消息rDelayedQueue.remove(delayMessage);//添加消息rDelayedQueue.offer(delayMessage, delayMessage.getDelayTime(), delayMessage.getTimeUnit());}/*** 移除指定队列中的消息*/public static void removeDelayMessage(DelayMessageDTO delayMessage) {log.info(取消delayMessage{}, delayMessage);if (!getDelayQueueMap().containsKey(delayMessage.getQueueName())) {log.error(queueName{},该延迟队列不存在请确认后再试..., delayMessage.getQueueName());return;}RDelayedQueueDelayMessageDTO rDelayedQueue getDelayQueueMap().get(delayMessage.getQueueName());rDelayedQueue.remove(delayMessage);removeDelayQueue(delayMessage);}/*** 从所有队列中删除消息*/public static void removeDelayQueue(DelayMessageDTO value) {DelayQueueEnum[] queueEnums DelayQueueEnum.values();for (DelayQueueEnum queueEnum : queueEnums) {RBlockingDequeObject blockingDeque getRedissonClient().getBlockingDeque(queueEnum.getCode());RDelayedQueueObject delayedQueue getRedissonClient().getDelayedQueue(blockingDeque);delayedQueue.remove(value);}}}
参考了大佬的博文 https://lhalcyon.com/delay-task/index.html
