针对客户端提交的合成视频任务,按照提交时间延迟60秒进行执行
1、使用redis的Java扩展库redisson提供的RBlocingQueue + RDelayDeque来实现
2、使用类似kafka时间轮的方式实现延迟队列
经过在本地测试发现,第一种实现起来工作量要小一些,易于操作,也易于理解;第二种借鉴kafka的时间轮方式,执行延迟任务时间更加精确,但是操作成本要高一些;如果对执行任务时间精度没有太高要求,可以直接选择第一种redisson的方式;
向RDelayedDueue队列存入要进行延迟处理的任务对象:
@Slf4j @Component public class SynthesisResultConsumer<T> { @Autowired private SynthesisResultComponent synthesisResultComponent; @Autowired private RDelayedQueue<SynthesisResultQryDto> delayedQueue; private static final int RETRY_MAX = 60; @Async public void accept(SynthesisResultQryDto qryDto) { qryDto.setRetryNum(qryDto.getRetryNum() + 1); SynthesisResultDto resultDto = synthesisResultComponent.getSynthesisResult(qryDto.getVideoId()); //合成结束 更新返回结果 if (resultDto.getIsEnd()) { synthesisResultComponent.updateSynthesisResult(resultDto); log.info("accept 查询合成视频结果 结果已出,videoId:{}", qryDto.getVideoId()); } else if (qryDto.getRetryNum() >= RETRY_MAX) { log.error("accept 查询合成视频结果 超过60次,请检查是否存异常,videoId:{}", qryDto.getVideoId()); } else { log.warn("accept 查询合成视频结果 本次查询未得到结果加入队列等待下次执行,videoId:{}", qryDto.getVideoId()); delayedQueue.offer(qryDto, 60, TimeUnit.SECONDS); } } }
从RBlockingQueue阻塞队列中取出到期的任务执行:
@Slf4j @Component public class SynthesisResultDelayJob { @Autowired private SynthesisResultConsumer<SynthesisResultQryDto> synthesisResultConsumer; @Autowired private RBlockingDeque<SynthesisResultQryDto> blockingDeque; @PostConstruct public void listen() { new Thread(() -> { while (true) { try { log.info("listen 本次监听时间:{}", DateUtil.date2String(new Date(), "yyyy-MM-dd HH:mm:ss")); SynthesisResultQryDto dto = blockingDeque.take(); log.info("listen 从队列中获取需要查询结果的合成视频任务信息:{}", JSON.toJSONString(dto)); synthesisResultConsumer.accept(dto); } catch (InterruptedException e) { log.error("listen InterruptedException,error msg:{}", ExceptionUtils.getMessage(e)); } } }).start(); } }
配置类:
@Configuration public class SynthesisResultDelayConfig { @Autowired private RedissonClient redissonClient; @Bean("blockQueue") public RBlockingDeque<SynthesisResultQryDto> getBlockQueue() { RBlockingDeque<SynthesisResultQryDto> blockingDeque = redissonClient.getBlockingDeque("prompter:synthesis:task:result"); return blockingDeque; } @Bean("delayedQueue") public RDelayedQueue<SynthesisResultQryDto> getDelayQueue() { RBlockingDeque<SynthesisResultQryDto> blockingDeque = getBlockQueue(); RDelayedQueue<SynthesisResultQryDto> delayedQueue = redissonClient.getDelayedQueue(blockingDeque); return delayedQueue; } }
依赖:
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> <version>3.11.4</version> </dependency>
上面的示例可以看出,这里的延迟队列使用了两个队列,一个是RBlockingQueue,这个队列是用来取延迟任务的;一个是RDelayedDeque,这个队列是用来存延迟任务的;offer方法中第二个参数表示任务延迟时间;
横向比较kafka时间轮的实现方案和redisson的延迟队列的实现方案,可以充分理解不同方案的优劣和使用场景,选择最终的方案要综合考虑实现成本和使用场景,最终选定最优的方案