java - 使用错误管理在 Java 中运行多线程

Question

我有一个像这样的工作接口(interface) IJob:

public interface IJob {
    public void execute();
}

在我的应用程序中，我有多个实现此接口(interface)的类，例如 IJob1 和 IJob2:

public class IJob1 implements IJob{
@Override
public void execute(){
    System.out.println("IJob1\n");
    }
}

public class IJob2 implements IJob{
@Override
public void execute(){
    System.out.println("IJob2\n");
    }
}

由于要运行的作业数量稳步增加，我想创建一个新作业，它获取 IJob 实例列表并并行运行它们。新实现用于并行运行作业的线程数量应该是可配置的。如果其中一个作业抛出异常，则所有其他当前正在运行的作业也应停止，并且 execute() 方法应将异常传递给调用者。

我写了这个，但我无法运行作业并检查是否有错误:

import java.util.LinkedList;

public class WorkQueue
{
    private final int nThreads;
    private final IJob[] threads;
    private final LinkedList queue;

    public WorkQueue(int nThreads)
    {
        this.nThreads = nThreads;
        queue = new LinkedList();
        threads = new IJob[nThreads];

        for (int i=0; i<nThreads; i++) {
            threads[i] = new IJob();
            threads[i].execute();
        }
    }

    public void execute(Runnable r) {
    synchronized(queue) {
        queue.addLast(r);
        queue.notify();
    }
}

private class PoolWorker extends Thread {
    public void run() {
        Runnable r;

        while (true) {
            synchronized(queue) {
                while (queue.isEmpty()) {
                    try
                    {
                        queue.wait();
                    }
                    catch (InterruptedException ignored)
                    {
                    }
                }

                r = (Runnable) queue.removeFirst();
            }

            // If we don't catch RuntimeException, 
            // the pool could leak threads
            try {
                r.run();
            }
            catch (RuntimeException e) {
                // You might want to log something here
            }
        }
    }
}
}

你能给我一些帮助吗？非常感谢。

Answer 1

我会推荐托管线程池。一个典型的模式是使用 Java Executors获得ExecutorService . ExecutorService 通常使用线程池和作业队列来实现。在 ExecutorService 上，有像 shutdownNow() 这样的方法，它“试图停止所有正在执行的任务”。这听起来像你想要做的。

例子，

List<Callable<Result>> tasks = new ArrayList<>();
for (final Object job: jobs) {
    final Callable<Result> task = new Callable<Result>() {
            @Override
            public Result call() throws Exception {
                // Do job here
                return result;
            }
        };
    tasks.add(task);
}
final numOfThreads = 20; 
final ExecutorService executor = Executors.newFixedThreadPool(numOfThreads);
try {
    executor.invokeAll(tasks);
} finally {
    executor.shutdownNow();
}