java - 同步线程池ExecutorService

Question

这是我第一次使用线程池，我不太了解executorservice的工作方式。我将水印放在图像上，然后将它们合并到一张空白图片上。但是即使我只使用一个线程，它也只会画一半。

这是我的 WorkerThread类:

public class WorkerThread implements Runnable {

    BufferedImage source;
    BufferedImage toDraw;
    int x;
    int y;
    BufferedImage target;
    ParallelWatermarkFilter pf;

    public WorkerThread(BufferedImage source, BufferedImage toDraw, int x, int y, BufferedImage target){
        this.source = source;
        this.toDraw = toDraw;
        this.x = x;
        this.y = y;
        this.target = target;
        pf = new ParallelWatermarkFilter(source, 5);
    }

    @Override
    public void run() {
        pf.mergeImages(source, toDraw, x, y, target);
    }
}

这就是我在 FilterClass 中使用ExecutorService的方式:

    public BufferedImage apply(BufferedImage input) {

        ExecutorService threadpool = Executors.newFixedThreadPool(numThreads);

                for (int w = 0; w < imgWidth; w += watermarkWidth) {
      for (int h = 0; h < imgHeight; h += watermarkHeight) {
            Runnable worker = new WorkerThread(input, watermark, w, h, result);
            System.out.println("WIDTH: " + w + "   HEIGHT: " + h);
            threadpool.execute(worker);
      }
    }

    threadpool.shutdown();

线程是否不等到一个线程完成？

Answer 1

ThreadPoolExecutor关闭和任务执行/排空工作队列/从工作队列中取出是一件很容易的事。因此，您不能依赖线程中断机制或其他方法。您所能保证的是:

Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted. Invocation has no additional effect if already shut down.

This method does not wait for previously submitted tasks to complete execution.

为了更深入地了解ThreadPoolExecutor实现，让我们看一下主要的执行方法:

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

这里的关键部分是调用getTask()。其片段是:

 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
     decrementWorkerCount();
     return null;
 }

该方法不同步，仅依赖于CAS'ed ctl值提供的排序。此处的ctl是AtomicInteger内部存储的全局池状态(用于非阻塞原子ThreadPoolExecutor状态获取)。

因此，以下情况是可能的。

称为getTask的工作线程

worker 线程获取池的运行状态。它仍然是RUNNING。

另一个线程启动了订单关闭操作，并相应地修改了ctl。

Worker线程已经从工作队列中接收了一项任务。