Redis实现延迟队列方法介绍

/**
 * 消息结构
 *
 * @author 睁眼看世界
 * @date 2020年1月15日
 */
@Data
public class Job implements Serializable {
 
 private static final long serialVersionUID = 1L;
 
 /**
 * Job的唯一标识。用来检索和删除指定的Job信息
 */
 @NotBlank
 private String jobId;
 
 
 /**
 * Job类型。可以理解成具体的业务名称
 */
 @NotBlank
 private String topic;
 
 /**
 * Job需要延迟的时间。单位：秒。（服务端会将其转换为绝对时间）
 */
 private Long delay;
 
 /**
 * Job的内容，供消费者做具体的业务处理，以json格式存储
 */
 @NotBlank
 private String body;
 
 /**
 * 失败重试次数
 */
 private int retry = 0;
 
 /**
 * 通知URL
 */
 @NotBlank
 private String url;
}
4.2.2 Job删除对象
/**
 * 消息结构
 *
 * @author 睁眼看世界
 * @date 2020年1月15日
 */
@Data
public class JobDie implements Serializable {
 
 private static final long serialVersionUID = 1L;
 
 /**
 * Job的唯一标识。用来检索和删除指定的Job信息
 */
 @NotBlank
 private String jobId;
 
 
 /**
 * Job类型。可以理解成具体的业务名称
 */
 @NotBlank
 private String topic;
}

/**
 * 搬运线程
 *
 * @author 睁眼看世界
 * @date 2020年1月17日
 */
@Slf4j
@Component
public class CarryJobScheduled {
 
 @Autowired
 private RedissonClient redissonClient;
 
 /**
 * 启动定时开启搬运JOB信息
 */
 @Scheduled(cron = "*/1 * * * * *")
 public void carryJobToQueue() {
 System.out.println("carryJobToQueue --->");
 RLock lock = redissonClient.getLock(RedisQueueKey.CARRY_THREAD_LOCK);
 try {
 boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
 }
 RScoredSortedSet<Object> bucketSet = redissonClient.getScoredSortedSet(RD_ZSET_BUCKET_PRE);
 long now = System.currentTimeMillis();
 Collection<Object> jobCollection = bucketSet.valueRange(0, false, now, true);
 List<String> jobList = jobCollection.stream().map(String::valueOf).collect(Collectors.toList());
 RList<String> readyQueue = redissonClient.getList(RD_LIST_TOPIC_PRE);
 readyQueue.addAll(jobList);
 bucketSet.removeAllAsync(jobList);
 } catch (InterruptedException e) {
 log.error("carryJobToQueue error", e);
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }
}

@Slf4j
@Component
public class ReadyQueueContext {
 
 @Autowired
 private RedissonClient redissonClient;
 
 @Autowired
 private ConsumerService consumerService;
 
 /**
 * TOPIC消费线程
 */
 @PostConstruct
 public void startTopicConsumer() {
 TaskManager.doTask(this::runTopicThreads, "开启TOPIC消费线程");
 }
 
 /**
 * 开启TOPIC消费线程
 * 将所有可能出现的异常全部catch住，确保While(true)能够不中断
 */
 @SuppressWarnings("InfiniteLoopStatement")
 private void runTopicThreads() {
 while (true) {
 RLock lock = null;
 try {
 lock = redissonClient.getLock(CONSUMER_TOPIC_LOCK);
 } catch (Exception e) {
 log.error("runTopicThreads getLock error", e);
 }
 try {
 if (lock == null) {
 continue;
 }
 // 分布式锁时间比Blpop阻塞时间多1S，避免出现释放锁的时候，锁已经超时释放，unlock报错
 boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 continue;
 }
 
 // 1. 获取ReadyQueue中待消费的数据
 RBlockingQueue<String> queue = redissonClient.getBlockingQueue(RD_LIST_TOPIC_PRE);
 String topicId = queue.poll(60, TimeUnit.SECONDS);
 if (StringUtils.isEmpty(topicId)) {
 continue;
 }
 
 // 2. 获取job元信息内容
 RMap<String, Job> jobPoolMap = redissonClient.getMap(JOB_POOL_KEY);
 Job job = jobPoolMap.get(topicId);
 
 // 3. 消费
 FutureTask<Boolean> taskResult = TaskManager.doFutureTask(() -> consumerService.consumerMessage(job.getUrl(), job.getBody()), job.getTopic() + "-->消费JobId-->" + job.getJobId());
 if (taskResult.get()) {
 // 3.1 消费成功，删除JobPool和DelayBucket的job信息
 jobPoolMap.remove(topicId);
 } else {
 int retrySum = job.getRetry() + 1;
 // 3.2 消费失败，则根据策略重新加入Bucket
 
 // 如果重试次数大于5，则将jobPool中的数据删除，持久化到DB
 if (retrySum > RetryStrategyEnum.RETRY_FIVE.getRetry()) {
 jobPoolMap.remove(topicId);
 continue;
 }
 job.setRetry(retrySum);
 long nextTime = job.getDelay() + RetryStrategyEnum.getDelayTime(job.getRetry()) * 1000;
 log.info("next retryTime is [{}]", DateUtil.long2Str(nextTime));
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.add(nextTime, topicId);
 // 3.3 更新元信息失败次数
 jobPoolMap.put(topicId, job);
 }
 } catch (Exception e) {
 log.error("runTopicThreads error", e);
 } finally {
 if (lock != null) {
 try {
 lock.unlock();
 } catch (Exception e) {
 log.error("runTopicThreads unlock error", e);
 }
 }
 }
 }
 }
}

/**
 * 提供给外部服务的操作接口
 *
 * @author why
 * @date 2020年1月15日
 */
@Slf4j
@Service
public class RedisDelayQueueServiceImpl implements RedisDelayQueueService {
 
 @Autowired
 private RedissonClient redissonClient;
 
 
 /**
 * 添加job元信息
 *
 * @param job 元信息
 */
 @Override
 public void addJob(Job job) {
 
 RLock lock = redissonClient.getLock(ADD_JOB_LOCK + job.getJobId());
 try {
 boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
 }
 String topicId = RedisQueueKey.getTopicId(job.getTopic(), job.getJobId());
 
 // 1. 将job添加到 JobPool中
 RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
 if (jobPool.get(topicId) != null) {
 throw new BusinessException(ErrorMessageEnum.JOB_ALREADY_EXIST);
 }
 
 jobPool.put(topicId, job);
 
 // 2. 将job添加到 DelayBucket中
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.add(job.getDelay(), topicId);
 } catch (InterruptedException e) {
 log.error("addJob error", e);
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }
 
 
 /**
 * 删除job信息
 *
 * @param job 元信息
 */
 @Override
 public void deleteJob(JobDie jobDie) {
 
 RLock lock = redissonClient.getLock(DELETE_JOB_LOCK + jobDie.getJobId());
 try {
 boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL);
 }
 String topicId = RedisQueueKey.getTopicId(jobDie.getTopic(), jobDie.getJobId());
 
 RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
 jobPool.remove(topicId);
 
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.remove(topicId);
 } catch (InterruptedException e) {
 log.error("addJob error", e);
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }
}