如何实现阻塞线程池执行器？

Question

我们有一个大文本文件，其中每一行都需要大量的process。设计是有一个class，它读取文件并将每一行的处理委托给thread，通过thread pool。一旦池中没有空闲线程来进行处理，应该阻止文件读取器类读取下一行。所以我需要一个blocking thread pool

在当前的实现中，ThreadPoolExecutor.submit() 和 ThreadPoolExecutor.execute() 方法在配置的线程数变忙后抛出 RejectedExecutionException 异常，如下面的代码 sn-p 所示。

public class BlockingTp {

    public static void main(String[] args) {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(3);
        ThreadPoolExecutor executorService=
            new ThreadPoolExecutor(1, 3, 30, TimeUnit.SECONDS, blockingQueue);
        int Jobs = 10;
        System.out.println("Starting application with " + Jobs + " jobs");
        for (int i = 1; i <= Jobs; i++)
            try {
                executorService.submit(new WorkerThread(i));
                System.out.println("job added " + (i));
            } catch (RejectedExecutionException e) {
                System.err.println("RejectedExecutionException");
            }
    }
}

class WorkerThread implements Runnable {
    int job;
    public WorkerThread(int job) {
        this.job = job;
    }
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (Exception excep) {
        }
    }
}

上述程序的输出是

Starting application to add 10 jobs
Added job #1
Added job #2
Added job #3
Added job #4
Added job #5
Added job #6
RejectedExecutionException
RejectedExecutionException
RejectedExecutionException
RejectedExecutionException

有人可以说明一下，即我如何实现阻塞线程池。

Answer 1

有人能说明一下吗，即我如何实现阻塞线程池。

您需要在执行器服务上设置拒绝执行处理程序。当线程去将作业放入执行器时，它会阻塞，直到阻塞队列中有空间。

BlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(3);
ThreadPoolExecutor executorService =
     new ThreadPoolExecutor(1, 3, 30, TimeUnit.SECONDS, arrayBlockingQueue);
// when the blocking queue is full, this tries to put into the queue which blocks
executorService.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            // block until there's room
            executor.getQueue().put(r);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RejectedExecutionException("Producer interrupted", e);
        }
    }
});

因此，TRE 不会抛出 RejectedExecutionException，而是调用拒绝处理程序，该处理程序反过来会尝试将作业放回队列中。这会阻止调用者。

Answer 2

让我们再看看你的代码：

for (int i = 1; i <= Jobs; i++)
  try {
    tpExe.submit(new WorkerThread(i));
    System.out.println("job added " + (i));
  } catch (RejectedExecutionException e) {
    System.err.println("RejectedExecutionException");
  }

所以 - 当您尝试提交并且池很忙时，就会引发该异常。如果你想绕过它，它可能看起来像：

public void yourSubmit(Runnable whatever) {
  boolean submitted = false;
  while (! submitted ) {
    try {
      tpExe.submit(new WorkerThread(whatever));
      submitted = true;
    } catch (RejectedExecutionException re) {
      // all threads busy ... so wait some time
      Thread.sleep(1000);
    }

换句话说：将该异常用作当前无法提交的“标记”。

Answer 3

这对我有用。

class handler implements RejectedExecutionHandler{
    @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                try {
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

            }

Answer 4

您可以使用信号量来控制资源。读取器将通过获取信号量来读取并创建异步任务。如果每个线程都忙，读取器线程将等待直到线程可用。

public class MyExecutor {
    private final Executor exec;
    private final Semaphore semaphore;

    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable command)
            throws InterruptedException, RejectedExecutionException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
            throw e;
        }
    }
}