主页 / 问题 / 1251134
Accepted
Василий Никпуп
Asked:2022-03-03 22:04:16 +0000 UTC2022-03-03 22:04:16 +0000 UTC

当池已满时如何阻塞线程？

  • 772

我有这个测试代码：

    ExecutorService pool = Executors.newFixedThreadPool(1);

    System.out.println("Starting adding new messages to pool");
    for (int i = 0; i < 60; i++) {
        pool.submit(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
                System.out.println("Done!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    System.out.println("End of the program");
    Thread.sleep(10_000);

如果我运行它，我会看到：

Starting adding new messages to pool
End of the program
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!

但是当缓冲区已满时，我需要让线程池生产者阻塞，然后在控制台中我会得到：

Starting adding new messages to pool
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!

我怎样才能做到这一点？

java

2 个回答

  1. Best Answer
    // завершаем работу ExecutorService (ExecutorService продолжит обрабатывать оставшийся список задач, но нельзя будет добавить новые).
pool.shutdown();
// блокируем текущий поток в ожидании завершения задач или таймаута, что произойдет раньше
pool.awaitTermination(600, TimeUnit.SECONDS);

    添加：

    如果要在池已满时阻止 for 循环，则应使用信号量：

    final int threads = 1;
// сделаем семафор на threads разрешений
final Semaphore lock = new Semaphore(threads);
ExecutorService pool =  Executors.newFixedThreadPool(threads);

System.out.println("Starting adding new messages to pool");
for (int i = 0; i < 60; i++) {
    // запрашиваем разрешение, acquire() грубо говоря повышает счетчик в семафоре на +1, и если этот счетчик равен threads, то метод заблокирует поток, пока счетчик не станет меньше threads (а это произойдет при вызове release())
    lock.acquire();

    pool.submit(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(100);
            System.out.println(new Date() + " Done! ");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            finally {
                lock.release();
            }
        });


        // небольшой костылёк, который реализует ожидание последних threads (кол-во) задач 
        if (i == 60 - 1) {
            lock.acquire(threads);
        }
    }

    lock.release(threads);

    System.out.println("End of the program");
    Thread.sleep(10_000);
    • 1

  2. 借助Java，无需自行编写自行车

    UPD不是没有自写的自行车，而是自行车的尺寸比有信号量的小。

    // велосипед
private static boolean interruptibleInfiniteOffer(BlockingQueue<Runnable> q, Runnable r) {
    try {
        return q.offer(r, Long.MAX_VALUE, TimeUnit.NANOSECONDS); // infinite == ~292 years
    } catch (InterruptedException e) {
        return false;
    }
}

public static ThreadPoolExecutor getFixedSizePoolWithLimitedWaitingQueue(int nThreads, int maxItemsInTheQueue) {
    BlockingQueue<Runnable> queue = maxItemsInTheQueue == 0
            ? new SynchronousQueue<>() { public boolean offer(Runnable r) { return interruptibleInfiniteOffer(this, r);} }
            : new ArrayBlockingQueue<>(maxItemsInTheQueue) { public boolean offer(Runnable r) { return interruptibleInfiniteOffer(this, r);} };
    return new ThreadPoolExecutor(nThreads, nThreads, 0, TimeUnit.MILLISECONDS, queue);
}

public static void main(String[] args) throws InterruptedException {
    int nThreads = 4;
    int maxItemsInTheQueue = 1;
    ThreadPoolExecutor tpe = getFixedSizePoolWithLimitedWaitingQueue(nThreads, maxItemsInTheQueue);

    System.out.println("Starting adding new messages to pool");

    final Random random = new Random();
    for (int i = 0; i < 60; i++) {
        final int t = i;

        System.out.println("Adding task "+t+"...");
        tpe.submit(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(50 + random.nextInt(50));
                System.out.println("Task "+t+" is completed");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println("Task "+t+" is added");
    };

    tpe.shutdown();
    System.out.println("End of the program");
}

    借助参数maxItemsInTheQueue，您可以控制队列中可以等待的元素数量。如果为 0，我们使用SynchronousQueue（任务不进入队列，生产者正在等待空闲线程），如果更多 - 然后ArrayBlockingQueue（任务进入队列直到队列满，然后生产者开始等待空闲线程）。

    此外，您可以offer(r, Long.MAX_VALUE, TimeUnit.NANOSECONDS)对其进行配置，以便等待队列中的空闲槽不会是无限的，而是会超时。

    • 0

相关问题