等到所有线程完成他们在java中的工作

Question

我正在编写一个包含5个线程的应用程序,它们可以同时从Web获取一些信息,并在缓冲类中填充5个不同的字段.
当所有线程完成工作时,我需要验证缓冲区数据并将其存储在数据库中.
我该怎么做(当所有线程完成工作时收到警报)？

Answer 1

我采用的方法是使用ExecutorService来管理线程池.

ExecutorService es = Executors.newCachedThreadPool();
for(int i=0;i<5;i++)
    es.execute(new Runnable() { /*  your task */ });
es.shutdown();
boolean finished = es.awaitTermination(1, TimeUnit.MINUTES);
// all tasks have finished or the time has been reached.

Answer 2

你可以join到线程.连接阻塞直到线程完成.

for (Thread thread : threads) {
    thread.join();
}

注意join抛出一个InterruptedException.如果发生这种情况,您将不得不决定该怎么做(例如,尝试取消其他线程以防止不必要的工作).

Answer 3

看看各种解决方案.

1. join()API已在早期版本的Java中引入.自JDK 1.5发布以来,这个并发包提供了一些很好的替代方案.

2. ExecutorService中的invokeAll#()

   执行给定的任务,在完成所有操作后返回持有其状态和结果的Futures列表.

   有关代码示例,请参阅此相关SE问题:

   如何使用invokeAll()让所有线程池完成他们的任务？

3. CountDownLatch

   允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助.

   一个CountDownLatch初始化为给定数.由于方法的调用,await方法阻塞直到当前计数达到零countDown(),之后释放所有等待的线程,并且任何后续的await调用立即返回.这是一次性现象 - 计数无法重置.如果您需要重置计数的版本,请考虑使用CyclicBarrier.

   有关使用的信息,请参阅此问题 CountDownLatch

   如何等待一个产生它自己的线程的线程？

4. ForkJoinPool或 newWorkStealingPool()中执行人

5. 迭代提交后创建的所有Future对象ExecutorService

Answer 4

等待/阻塞 Thread Main，直到其他一些线程完成它们的工作。

如前所述，@Ravindra babu它可以通过各种方式实现，但通过示例进行展示。

• java.lang.Thread。加入（） ^自：1.0

  public static void joiningThreads() throws InterruptedException {
      Thread t1 = new Thread( new LatchTask(1, null), "T1" );
      Thread t2 = new Thread( new LatchTask(7, null), "T2" );
      Thread t3 = new Thread( new LatchTask(5, null), "T3" );
      Thread t4 = new Thread( new LatchTask(2, null), "T4" );
  
      // Start all the threads
      t1.start();
      t2.start();
      t3.start();
      t4.start();
  
      // Wait till all threads completes
      t1.join();
      t2.join();
      t3.join();
      t4.join();
  }
  

• java.util.concurrent.CountDownLatch ^{从 1.5 开始}

  • .countDown() « 减少锁存器组的计数。
  • .await() « await 方法会阻塞，直到当前计数达到零。

  如果你创建了latchGroupCount = 4那么countDown()应该叫4倍，使数0，所以，这await()将释放阻塞线程。

  public static void latchThreads() throws InterruptedException {
      int latchGroupCount = 4;
      CountDownLatch latch = new CountDownLatch(latchGroupCount);
      Thread t1 = new Thread( new LatchTask(1, latch), "T1" );
      Thread t2 = new Thread( new LatchTask(7, latch), "T2" );
      Thread t3 = new Thread( new LatchTask(5, latch), "T3" );
      Thread t4 = new Thread( new LatchTask(2, latch), "T4" );
  
      t1.start();
      t2.start();
      t3.start();
      t4.start();
  
      //latch.countDown();
  
      latch.await(); // block until latchGroupCount is 0.
  }
  

Threaded 类的示例代码LatchTask。测试方法使用joiningThreads(); 和latchThreads();来自主要方法。

class LatchTask extends Thread {
    CountDownLatch latch;
    int iterations = 10;
    public LatchTask(int iterations, CountDownLatch latch) {
        this.iterations = iterations;
        this.latch = latch;
    }

    @Override
    public void run() {
        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + " : Started Task...");

        for (int i = 0; i < iterations; i++) {
            System.out.println(threadName + " : " + i);
            MainThread_Wait_TillWorkerThreadsComplete.sleep(1);
        }
        System.out.println(threadName + " : Completed Task");
        // countDown() « Decrements the count of the latch group.
        if(latch != null)
            latch.countDown();
    }
}

• CyclicBarriers一种同步辅助工具，它允许一组线程相互等待以到达公共屏障点。CyclicBarriers 在涉及固定大小的线程方的程序中很有用，这些线程偶尔必须相互等待。屏障被称为循环的，因为它可以在等待线程被释放后重新使用。

  CyclicBarrier barrier = new CyclicBarrier(3);
  barrier.await();
  

  例如参考这个Concurrent_ParallelNotifyies类。

• Executer 框架：我们可以使用ExecutorService创建线程池，并使用Future 跟踪异步任务的进度。

  • submit(Runnable)，submit(Callable)返回未来对象。通过使用future.get()函数，我们可以阻塞主线程，直到工作线程完成其工作。

  • invokeAll(...) - 返回一个 Future 对象列表，通过它您可以获得每个 Callable 的执行结果。

查找使用 Interfaces Runnable、Callable with Executor 框架的示例。

@也可以看看

• 找出线程还活着？

Answer 5

除了Thread.join()其他人的建议外,java 5还引入了执行程序框架.在那里你不使用Thread对象.相反,您将您的Callable或Runnable对象提交给执行者.有一个特殊的执行程序,用于执行多个任务并无序返回结果.那就是ExecutorCompletionService:

ExecutorCompletionService executor;
for (..) {
    executor.submit(Executors.callable(yourRunnable));
}

然后你可以重复调用,take()直到没有更多的Future<?>对象返回,这意味着所有对象都已完成.

根据您的情况,可能相关的另一件事是CyclicBarrier.

一种同步辅助工具,允许一组线程全部等待彼此到达公共障碍点.CyclicBarriers在涉及固定大小的线程方的程序中很有用,这些线程必须偶尔等待彼此.屏障称为循环,因为它可以在等待线程释放后重新使用.

Answer 6

另一种可能性是CountDownLatch对象,它对于简单情况很有用:因为您事先知道线程数,所以使用相关计数初始化它,并将对象的引用传递给每个线程.
完成任务后,每个线程调用CountDownLatch.countDown()内部计数器递减.主线程在启动所有其他线程后,应该执行CountDownLatch.await()阻塞调用.一旦内部计数器达到0,它就会被释放.

注意使用此对象,InterruptedException也可以抛出一个.

Answer 7

你做

for (Thread t : new Thread[] { th1, th2, th3, th4, th5 })
    t.join()

在此for循环之后,您可以确保所有线程都已完成其作业.