java - 如何在任何一个线程完成后终止所有其他正在运行的线程

Question

问题:我有一组线程在循环中并行启动。首先退出任何线程后，必须终止所有其他正在运行的线程。这是我尝试过的，但它不起作用。任何帮助表示赞赏。

public class ThreadsMain {

    public static void main(String[] args) {
        int SIZE = 3;
        Thread t[] = new Thread[SIZE];

        for (int i = 0; i < SIZE; i++) {
            myThreads th = new myThreads();
            t[i] = new Thread(th);
            t[i].start();

        }

    }
}

Answer 1

这是一种方法，使用内部锁实现同步器，并使用中断来取消未完成的任务。该数据结构使消费者线程阻塞，直到生产者提交结果，然后取消其他工作线程。

这是一个玩具示例，请参阅末尾的链接以了解实际的方法。

首先，这是一个接受结果的线程安全数据结构，它允许线程注册为监听器，并在有结果提交给它时中断它们:

class MyQueue<T> {
    private java.util.List<T> results = new java.util.ArrayList<T>();
    private java.util.List<Thread> listeners = new java.util.ArrayList<Thread>();

    public synchronized void put(T o) {
        results.add(o);
        notifyAll();
        for (Thread listener : listeners) {
            listener.interrupt();
        }
    }

    public synchronized T take() throws InterruptedException {
        while (results.size() == 0) {
            wait();            
        }
        return results.remove(0);
    }

    public synchronized void addListener(Thread t) {
        listeners.add(t);
    }
}

(我不喜欢让这个类对听众了解太多，但我也不想过度考虑一个玩具示例。)

wait 方法释放锁并使调用线程进入 hibernate 状态，直到发生通知(或者它可以任意停止等待)。它使用结果列表的大小属性来了解结果何时提交。假设因为线程停止等待而可以推断出有关当前状态的某些信息是不安全的，一旦线程重新获取锁，它就需要检查当前状态的实际情况。有关等待如何工作的更多信息，请参阅 this tutorial .

这是一个计算结果的任务(在迭代之间 hibernate ，以便这些线程可以运行一段时间):

class FibTask implements Runnable {

    private final MyQueue<BigInteger> queue;
    private final int n;
    private long sleepTime;

    public FibTask(int n, long sleepTime, MyQueue<BigInteger> queue) {
        this.n = n;
        this.sleepTime = sleepTime;
        this.queue = queue;
    }

    @Override public void run() {
        BigInteger a = BigInteger.valueOf(0);
        BigInteger b = BigInteger.valueOf(1);
        int i = 0;
        try {
            while (!Thread.currentThread().isInterrupted() && i < n) {
                i = i + 1;
                BigInteger temp = a;
                a = b;
                b = a.add(temp);
                Thread.sleep(sleepTime);
            }    
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        if (!Thread.currentThread().isInterrupted()) {
            queue.put(b);
        }
    }
}

请注意上面代码中的 Runnable 如何需要知道中断它的尝试。中断是协作性的，任务负责决定何时检测中断并处理终止过程。

此外，如果任务涉及 IO，那么在某些情况下中断不起作用，您必须关闭套接字，请参阅 this article对此进行更多讨论。

这是运行线程并获取结果的主程序。 MyQueue 类已经完成了大部分工作，因此不必做太多工作:

class Completion {
    public static void main(String ... args) throws Exception {
        MyQueue<BigInteger> queue = new MyQueue<BigInteger>();
        Thread t1 = new Thread(new FibTask(10, 1000L, queue));
        Thread t2 = new Thread(new FibTask(20, 10000L, queue));
        Thread t3 = new Thread(new FibTask(25, 50000L, queue));
        queue.addListener(t1);
        queue.addListener(t2);
        queue.addListener(t3);
        t1.start();
        t2.start();
        t3.start();
        System.out.println(queue.take());
    }
}

请注意，这不是一场公平的竞赛，因为线程的启动是交错的，后面的线程处于劣势。将任务提交给预先初始化线程池的执行器将确保启动线程的时间不会导致延迟。

有关使用 Executors 和 Futures 等 java.util.concurrent 功能的更好方法，请参阅 API documentation for ExecutorCompletionService 中给出的示例。 .