等待所有线程完成java中的工作

Question

我正在编写一个具有 5 个线程的应用程序，这些线程同时从 Web 获取一些信息并填充缓冲区类中的 5 个不同字段。
当所有线程完成其工作时，我需要验证缓冲区数据并将其存储在数据库中。
我该如何做到这一点（当所有线程完成工作时收到警报）？

Answer 1

我创建了一个小辅助方法来等待几个线程完成：

public static void waitForThreadsToFinish(Thread... threads) {
        try {
            for (Thread thread : threads) {
                thread.join();
            }
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

Answer 2

您可以使用 Threadf#join< /a> 用于此目的的方法。

Answer 3

虽然与OP的问题无关，但如果您对只有一个线程的同步（更准确地说，是一次交会）感兴趣，您可以使用 Exchanger

在我的例子中，我需要暂停父线程，直到子线程执行某些操作，例如完成其初始化。 CountDownLatch 也可以很好地工作。

Answer 4

执行器服务可用于管理多个线程，包括状态和完成。请参阅http://programmingexamples.wikidot.com/executorservice

Answer 5

试试这个，会起作用的。

  Thread[] threads = new Thread[10];

  List<Thread> allThreads = new ArrayList<Thread>();

  for(Thread thread : threads){

        if(null != thread){

              if(thread.isAlive()){

                    allThreads.add(thread);

              }

        }

  }

  while(!allThreads.isEmpty()){

        Iterator<Thread> ite = allThreads.iterator();

        while(ite.hasNext()){

              Thread thread = ite.next();

              if(!thread.isAlive()){

                   ite.remove();
              }

        }

   }

Answer 6

我遇到了类似的问题，最终使用了 Java 8 parallelStream。

requestList.parallelStream().forEach(req -> makeRequest(req));

它非常简单且可读。
它在幕后使用默认 JVM 的 fork join 池，这意味着它将等待所有线程完成后再继续。对于我的情况来说，这是一个巧妙的解决方案，因为它是我的应用程序中唯一的并行流。如果您有多个并行流同时运行，请阅读下面的链接。

有关并行流的更多信息请参见此处。

Answer 7

现有的答案说可以 join() 每个线程。

但是有几种方法可以获取线程数组/列表：

• 在创建时将线程添加到列表中。
• 使用ThreadGroup 来管理线程。

以下代码将使用 ThreadGruop 方法。它首先创建一个组，然后在创建每个线程时在构造函数中指定该组，然后可以通过 ThreadGroup.enumerate() 获取线程数组

---

代码

SyncBlockLearn.java

import org.testng.Assert;
import org.testng.annotations.Test;

/**
 * synchronized block - learn,
 *
 * @author eric
 * @date Apr 20, 2015 1:37:11 PM
 */
public class SyncBlockLearn {
    private static final int TD_COUNT = 5; // thread count
    private static final int ROUND_PER_THREAD = 100; // round for each thread,
    private static final long INC_DELAY = 10; // delay of each increase,

    // sync block test,
    @Test
    public void syncBlockTest() throws InterruptedException {
        Counter ct = new Counter();
        ThreadGroup tg = new ThreadGroup("runner");

        for (int i = 0; i < TD_COUNT; i++) {
            new Thread(tg, ct, "t-" + i).start();
        }

        Thread[] tArr = new Thread[TD_COUNT];
        tg.enumerate(tArr); // get threads,

        // wait all runner to finish,
        for (Thread t : tArr) {
            t.join();
        }

        System.out.printf("\nfinal count: %d\n", ct.getCount());
        Assert.assertEquals(ct.getCount(), TD_COUNT * ROUND_PER_THREAD);
    }

    static class Counter implements Runnable {
        private final Object lkOn = new Object(); // the object to lock on,
        private int count = 0;

        @Override
        public void run() {
            System.out.printf("[%s] begin\n", Thread.currentThread().getName());

            for (int i = 0; i < ROUND_PER_THREAD; i++) {
                synchronized (lkOn) {
                    System.out.printf("[%s] [%d] inc to: %d\n", Thread.currentThread().getName(), i, ++count);
                }
                try {
                    Thread.sleep(INC_DELAY); // wait a while,
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            System.out.printf("[%s] end\n", Thread.currentThread().getName());
        }

        public int getCount() {
            return count;
        }
    }
}

主线程将等待组中的所有线程完成。

Answer 8

我遇到了类似的情况，我必须等到所有子线程完成其执行，然后只有我才能获得每个子线程的状态结果..因此我需要等到所有子线程完成。

下面是我的代码，我使用以下方法进行多线程处理：

 public static void main(String[] args) {
        List<RunnerPojo> testList = ExcelObject.getTestStepsList();//.parallelStream().collect(Collectors.toList());
        int threadCount = ConfigFileReader.getInstance().readConfig().getParallelThreadCount();
        System.out.println("Thread count is : =========  " + threadCount); // 5

        ExecutorService threadExecutor = new DriverScript().threadExecutor(testList, threadCount);


        boolean isProcessCompleted = waitUntilCondition(() -> threadExecutor.isTerminated()); // Here i used waitUntil condition 

        if (isProcessCompleted) {
            testList.forEach(x -> {
                System.out.println("Test Name: " + x.getTestCaseId());
                System.out.println("Test Status : " + x.getStatus());
                System.out.println("======= Test Steps ===== ");

                x.getTestStepsList().forEach(y -> {
                    System.out.println("Step Name: " + y.getDescription());
                    System.out.println("Test caseId : " + y.getTestCaseId());
                    System.out.println("Step Status: " + y.getResult());
                    System.out.println("\n ============ ==========");
                });
            });
        }

下面的方法用于使用并行处理分发列表，

// This method will split my list and run in a parallel process with mutliple threads
    private ExecutorService threadExecutor(List<RunnerPojo> testList, int threadSize) {
        ExecutorService exec = Executors.newFixedThreadPool(threadSize);
        testList.forEach(tests -> {
            exec.submit(() -> {
                driverScript(tests);
            });
        });
        exec.shutdown();
        return exec;
    }

这是我的等待方法：在这里您可以等待，直到 do while 循环中的条件满足。就我而言，我等待了一些最大超时。
这将继续检查，直到您的 threadExecutor.isTermminate() 为 true 且轮询周期为 5 秒。

    static boolean waitUntilCondition(Supplier<Boolean> function) {
        Double timer = 0.0;
        Double maxTimeOut = 20.0;

        boolean isFound;
        do {
            isFound = function.get();
            if (isFound) {
                break;
            } else {
                try {
                    Thread.sleep(5000); // Sleeping for 5 sec (main thread will sleep for 5 sec)
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                timer++;
                System.out.println("Waiting for condition to be true .. waited .." + timer * 5 + " sec.");
            }
        } while (timer < maxTimeOut + 1.0);

        return isFound;
    }

Answer 9

在主线程中使用它： while(!executor.isTermminate());
从执行程序服务启动所有线程后放置这行代码。这只会在执行器启动的所有线程完成后才启动主线程。确保调用 executor.shutdown();在上述循环之前。

Answer 10

我采取的方法是使用 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 11

您可以 加入< /code>到线程。连接会阻塞，直到线程完成。

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

请注意，join 会抛出 InterruptedException。您必须决定如果发生这种情况该怎么办（例如，尝试取消其他线程以防止完成不必要的工作）。

Answer 12

看看各种解决方案。

1. join() API 已在 Java 的早期版本中引入。此 并发 提供了一些不错的替代方案 自 JDK 1.5 发布以来的软件包。

2. ExecutorService#invokeAll()

执行给定的任务，当一切完成时返回保存其状态和结果的 Future 列表。

请参阅此相关的 SE 问题以获取代码示例：

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

3. CountDownLatch

一种同步辅助工具，允许一个或多个线程等待，直到其他线程中执行的一组操作完成。

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

请参阅此问题以了解 CountDownLatch

4. ForkJoinPool 或 newWorkStealingPool() 在 执行者

5. 迭代所有Future 提交到 ExecutorService

   后创建的对象

来源：来自 docs.oracle.com 的各种链接

Answer 13

等待/阻塞主线程，直到其他线程完成其工作。

正如@Ravindra babu所说，它可以通过多种方式实现，但通过示例进行展示。

• java.lang.Thread。 join() ^自：1.0

  public static void joinThreads() 抛出 InterruptedException {
      线程 t1 = new Thread( new LatchTask(1, null), "T1" );
      线程 t2 = new Thread( new LatchTask(7, null), "T2" );
      线程 t3 = new Thread( new LatchTask(5, null), "T3" );
      线程 t4 = new Thread( new LatchTask(2, null), "T4" );
  
      // 启动所有线程
      t1.start();
      t2.start();
      t3.start();
      t4.start();
  
      // 等待所有线程完成
      t1.join();
      t2.join();
      t3.join();
      t4.join();
  }
  

• java.util.concurrent.CountDownLatch ^自：1.5

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

  如果您创建了 latchGroupCount = 4，则应调用 countDown() 4 次以使计数为 0。因此，await()将释放阻塞线程。

  public static voidatchThreads() 抛出 InterruptedException {
      int 锁存组计数 = 4;
      CountDownLatch 闩锁 = new CountDownLatch(latchGroupCount);
      线程 t1 = new Thread( new LatchTask(1, 锁存器), "T1" );
      线程 t2 = new Thread( new LatchTask(7, 锁存器), "T2" );
      线程 t3 = new Thread( new LatchTask(5, 锁存器), "T3" );
      线程 t4 = new Thread( new LatchTask(2, 锁存器), "T4" );
  
      t1.start();
      t2.start();
      t3.start();
      t4.start();
  
      //latch.countDown();
  
      闩锁.await(); // 阻塞直到latchGroupCount为0。
  }
  

线程类LatchTask的示例代码。要测试该方法，请使用 joiningThreads();
和 main 方法中的 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 屏障 = new CyclicBarrier(3);
  屏障.await();
  

  例如，请参考此 Concurrent_ParallelNotifyies 类。

---

• 执行器框架：我们可以使用ExecutorService来创建一个线程池，并通过Future跟踪异步任务的进度。

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

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

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

---

@另请参阅

• 找出线程仍然存在？

Answer 14

除了其他人建议的 Thread.join() 之外，java 5 还引入了执行器框架。在那里你不使用Thread对象。相反，您可以将 Callable 或 Runnable 对象提交给执行程序。有一个特殊的执行器，用于执行多个任务并无序返回其结果。这是 ExecutorCompletionService< /a>：

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

然后可以重复调用take()，直到不再有Future对象返回，这意味着所有对象都已完成。

---

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

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

Answer 15

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

注意，使用这个对象，也可能抛出 InterruptedException。

Answer 16

在这个 for 循环之后

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

，您可以确定所有线程都已完成其工作。

Answer 17

将线程对象存储到某个集合（如列表或集合）中，然后在线程启动后循环遍历该集合并调用 join() 在线程上。