java - java中的自定义线程池实现

Question

出于学习目的，我正在尝试在 Java 中实现我自己的线程池。以下是我已经实现的。我对这个实现有几个问题:

1. 虽然我正在使用 BlockingQueue，就像内置 java 一样，执行器希望我们提供 Runnable 对象(通过执行方法)。但就我而言，我觉得我可以创建任何对象而不是 Runnable。那么，为什么 Java 执行程序期望 Runnable，我尝试查看源代码但仍无法弄清楚。

2. 这个原始实现还有什么问题吗？

请找到代码。

公共(public)类 CustomThreadPool {

private final BlockingQueue<Runnable> blockingQueue;
private final Worker[] workers;

public CustomThreadPool(final int numOfThreads) {
    blockingQueue = new LinkedBlockingQueue<>();
    workers = new Worker[numOfThreads];

    for (int i = 0; i < numOfThreads; i++) {
        workers[i] = new Worker();
        workers[i].start();
    }
}

public void execute(final Runnable task) {
    blockingQueue.add(task);
}

public void shutdownImmediately() {
    for (int i = 0; i < workers.length; i++) {
        workers[i].shutdownSignal = true;
        workers[i] = null;
    }
}

 private class Worker extends Thread {
    private Runnable taskToPerform = null;
    boolean shutdownSignal = false;

    @Override
    public void run() {
        while(true && !shutdownSignal) {
            taskToPerform = blockingQueue.poll();
            if (taskToPerform != null) {
                taskToPerform.run();
            }
            if(shutdownSignal) {
                break;
            }
        }
    }
}

public static void main(String[] args) throws Exception {
    final CustomThreadPool threadPool = new CustomThreadPool(5);
    for (int i = 0; i < 20; i++) {
        threadPool.execute(() -> System.out.println(Thread.currentThread().getName()));
    }
    Thread.sleep(1*1000);
    threadPool.shutdownImmediately();
}

Answer 1

1. Executor 期望 Runnable 或 Callable，因为它在运行您提交的任务时会调用这些接口(interface)的 run 或 call 方法。

2. 在您的实现中，您不使用 BlockingQueue 的阻塞方面。当您的队列中不存在任何任务时，您的线程池线程将在您的 while(true && !shutdownSignal) 循环中不断旋转(占用 cpu 时间)。因为 poll() 方法没有阻塞。这不是您在实现线程池时想要的东西。

您应该使用其中一种阻塞方法而不是 poll()。

您可以使用 poll(long timeout,TimeUnit unit)需要超时参数的方法。在这种情况下，如果您调用 shutdownImmediately 方法，而任何线程池线程都在等待此调用。他们将等待超时持续时间，并且轮询将向他们返回 null 他们将看到正在设置 shutdownSignal 并退出循环。如果你不想让他们等待超时，你也可以调用中断方法。

 // in run method
 try {
     taskToPerform = blockingQueue.poll(5, TimeUnit.SECONDS);
 } catch (InterruptedException e) {
     break; // this thread is interrupted. Time to leave this loop
 }

// in shutdownImmediately method 
workers[i].shutdownSignal = true;
// If You don't call interrupt, Threads will wait at max 5 sec for this case.
workers[i].interrupt(); 
workers[i] = null;

或者您可以使用 take()当队列中没有任何内容时阻塞的方法。但在这种情况下，当您调用 shutdownImmediately 方法时，您必须中断可能正在等待 take 方法调用的线程。否则他们会卡在 take() 调用处。

// in run method
try {
   taskToPerform = blockingQueue.take();
} catch (InterruptedException e) {
   break; // this thread is interrupted. Time to leave this loop
}

// in shutdownImmediately method 
workers[i].shutdownSignal = true;
workers[i].interrupt(); // this is crucial for this case
workers[i] = null;