redis zset 延迟合并任务处理

redis zset 延迟合并任务处理

@Autowired
    public RedisTemplate redisTemplate;
    
##1.发送端：在接口中收集任务ID,累计时间段之后，合并处理。
##redis zset 主键，任务ID(不重复)，时间戳    
String key = "任务分组名称";
//延时
redisTemplate.opsForZSet().add(key,taskId,System.currentTimeMillis());
System.out.println("加入任务， taskId: " + taskId + ", exeTime: " + System.currentTimeMillis() + ", 当前时间：" + LocalDateTime.now() +",key = " + key);


##2.接收端
@Configuration
public class ExecutorServiceConf {
    @Bean
    public ExecutorService executorService() {
        return new ThreadPoolExecutor(
                20,
                40,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                new ThreadFactoryBuilder().setNameFormat("myMergeTask-pool-%d").build(),
                new ThreadPoolExecutor.AbortPolicy());
    }
}


@Service
public class ConsumerTask {
    @Autowired
    public RedisTemplate redisTemplate;
    @Autowired
    ExecutorService executorService;
    @Value("${delayTime:2}")
    private Integer delayTime;
    
    //服务启动的时候加载
    @PostConstruct
    public void consumer() {
        //启动线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    String key = "任务分组名称";
                    //rangeByScore 主键,0,当前时间,0,偏移量1000
                    Set<Integer> taskIdSet = redisTemplate.opsForZSet().rangeByScore(key, 0, System.currentTimeMillis(), 0, 1000);
                    if (taskIdSet == null || taskIdSet.isEmpty()) {
                        logger.info("没有任务");
                    } else {
                        //1.查询任务ID关联的记录

                        //2.任务合并处理
                        
                        //3.处理成功后,清理redis
                        taskIdSet.forEach(id -> {
                            //4.根据taskId更新任务状态
                            
                            //5.根据taskId删除redis
                            long result = redisTemplate.opsForZSet().remove(key, id);
                            if (result == 1L) {
                                logger.info("从延时队列中获取到任务,taskId:" + id + ",当前时间：" + LocalDateTime.now());
                            }
                        });
                    }

                    try {
                        //间隔2分钟时间
                        TimeUnit.MINUTES.sleep(delayTime);
                    } catch (InterruptedException e) {
                        logger.error("定时扫描exception:",e);
                    }
                }
            }
        });
    }
    
}