模拟任务积压

news/2024/7/8 9:18:46 标签: java, 中间件

这个demo没有实质意义，只是简单模拟一下任务积压的场景。

java">    private static final ExecutorService PRODUCER = Executors.newFixedThreadPool(1);
    private static final ThreadPoolExecutor CONSUMER = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
    private static final LinkedBlockingQueue<Runnable> QUEUE = new LinkedBlockingQueue<>();
    public static void main(String[] args) throws InterruptedException {
        // 启动生产者线程
        produce();
        // 启动消费者线程
        consume();
    }

    private static void produce() {
        // 启动一个新的线程作为生产者
        PRODUCER.execute(() -> {
            // 使用AtomicInteger来管理任务编号，确保线程安全
            AtomicInteger i = new AtomicInteger();
            while (true) {
                // 将一个任务添加到队列中
                QUEUE.add(() -> {
                    // 模拟任务需要长时间执行
                    try {
                        Thread.sleep(8000);
                    } catch (InterruptedException e) {
                        // 将中断异常转换为运行时异常，以便抛出
                        throw new RuntimeException(e);
                    }
                    // 执行任务，并输出任务编号
                    log.info("执行任务 {}" , i.getAndIncrement());
                });
                // 模拟生产者添加任务的频率，间隔2秒
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    // 将中断异常转换为运行时异常，以便抛出
                    throw new RuntimeException(e);
                }
            }
        });
    }

    private static void consume() throws InterruptedException {
        // 无限循环，用于持续处理队列中的任务
        while (true){
            // 从队列中获取一个任务，该任务是待执行的Runnable对象
            Runnable task = QUEUE.take();
            // 获取队列当前的大小，用于监控队列的状态
            int size = QUEUE.size();
            // 输出队列当前的大小，用于日志记录或调试
            log.info("当前队列大小 {}",size);
            // 执行获取到的任务
            CONSUMER.execute(task);
            try {
                // 使当前线程休眠8秒，目的是控制任务处理的频率
                // 或者说，给队列填充新任务留下时间
                Thread.sleep(8000);
            } catch (InterruptedException e) {
                // 线程被中断时，抛出运行时异常
                // 这样做是为了让上层调用者能够处理中断情况
                throw new RuntimeException(e);
            }
        }
    }