DelayQueue 实现单机延迟任务处理

Bertha 。 2022-10-11 14:56 195阅读 0赞

# 前言 #

我们有一些项目需要单机获取延迟处理一些任务，延迟的时间并不长。也没有延迟队列支持，项目本身docker部署利用redis 实现又存在队列锁竞争问题，该资源也不是非常重要如果不小心丢失部分也无关紧要。当然也可以先存入数据库，项目启动将数据加载的内存中。

# 实践 #

1、配置线程池处理处理数据，可以根据自身内部数据多少配置线程池，不确定可以进行配置方式处理。

@Component
    public class ParkingConfig { 
    
    
    
    
        public static  DelayQueue<ImageDelay> queue = new DelayQueue<>();
    
        @Bean("getParkImage")
        public ExecutorService executor(@Autowired ApplicationContext ctx) { 
           // final int processor = Runtime.getRuntime().availableProcessors();
            ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 1, 10,
                    TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000));
            threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            threadPool.setThreadFactory(runnable -> { 
                Thread thread = new Thread(runnable);
                final String name = "getParkImage-" + thread.getName();
                thread.setName(name);
                return thread;
            });
            for (int i = 0; i < threadPool.getCorePoolSize(); i++) { 
                threadPool.execute(new GetImageTask(ctx.getBean(ParkingTurnoverService.class)));
            }
            return threadPool;
        }
    }

2、创建GetImageTask任务进行处理数据,这里用take 根据队列特性，如果队列为空直接阻塞线程，不需要做休眠处理。

@Slf4j
    public class GetImageTask  implements Runnable { 
    
        private ParkingTurnoverService parkingTurnoverService;
    
        public GetImageTask(ParkingTurnoverService parkingTurnoverService) { 
            this.parkingTurnoverService = parkingTurnoverService;
        }
    
        @Override
        public void run() { 
            log.info("get Image start");
            ImageDelay imageDelay = null;
            while (true) { 
                try { 
                    imageDelay = ParkingConfig.queue.take();
                    this.parkingTurnoverService.insertImage(imageDelay);
                } catch (Throwable e) { 
                    log.error("get image is fail", e);
                }
            }
        }
    }

3、队列中需要存储需要处理的数据，DelayQueue 提供一个getDelay(TimeUnit unit) 做延迟实践的处理。

@Data
    public class ImageDelay implements Delayed { 
    
        /* 触发时间*/
        private long time;
    
        private Long id;
    
        private ImageEnum imageEnum;
    
        private Integer num;
    
        public ImageDelay(Long id, ImageEnum imageEnum, long time, TimeUnit unit) { 
            this.id = id;
            this.imageEnum = imageEnum;
            this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0);
            this.num = 1;
        }
    
    
        @Override
        public long getDelay(TimeUnit unit) { 
            return time - System.currentTimeMillis();
        }
    
        @Override
        public int compareTo(Delayed o) { 
            ImageDelay item = (ImageDelay) o;
            long diff = this.time - item.time;
            if (diff <= 0) { 
                return -1;
            }else { 
                return 1;
            }
        }
    }

4、放入数据处理,等待GetImageTask 对其进行处理

ImageDelay imageDelay = new ImageDelay(parkingTurnover.getId(), ImageEnum.IN, 3, TimeUnit.SECONDS);
    ParkingConfig.queue.add(imageDelay);