如何等待线程工厂完成其所有任务的执行

Question

我正在使用Spring 4.3.8.RELEASE和Java 7。 我想创建一个线程工厂来帮助管理应用程序中的某些工作线程。 我这样声明我的线程工厂

<bean id="myprojectThreadFactory" class="org.springframework.scheduling.concurrent.CustomizableThreadFactory">
    <constructor-arg value="prefix-"/>
</bean>
<bean id="myprojectTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="threadFactory" ref="myprojectThreadFactory"/>
    <property name="corePoolSize" value="${myproject.core.thread.pool.size}" />
    <property name="maxPoolSize" value="${myproject.max.thread.pool.size}" />
</bean>

但是，我在线程上"加入"时遇到问题。 也就是说，我想等待所有工作完成，然后再继续执行某项任务，所以我有

    m_importEventsWorker.work();
    m_threadExecutor.shutdown();
    System.out.println("done.");

我的线程池是这样执行的

public void work(final MyWorkUnit pmyprojectOrg)
{
    final List<MyWorkUnit> allOrgs = new ArrayList<MyWorkUnit>();
    if (pmyprojectOrg != null)
    {
        processData(pmyprojectOrg.getmyprojectOrgId());
    } else { 
        allOrgs.addAll(m_myprojectSvc.findAllWithNonEmptyTokens());
        // Cue up threads to execute
        for (final MyWorkUnit myprojectOrg : allOrgs)
        {
            m_threadExecutor.execute(new Thread(new Runnable(){
                @Override
                public void run()
                {
                    System.out.println("started.");
                    processData(myprojectOrg.getmyprojectOrgId());
                }
            }));
        }   // for

然而，打印出来的是

done.
started.
started.

所以很明显，我不是在等待。 等待线程完成工作的正确方法是什么？

Answer 1

您可以使用执行器服务创建一个固定的线程池，并检查池大小是否为空：

ExecutorService executor = Executors.newFixedThreadPool(50);

如果使用此执行器运行任务，并使用 fixedRate 或 fixedDelay 定期检查线程池大小@Scheduled则可以查看它们是否已完成。

ThreadPoolExecutor poolInfo = (ThreadPoolExecutor) executor;
Integer activeTaskCount = poolInfo.getActiveCount();
if(activeTaskCount = 0) {
    //If it is 0, it means threads are waiting for tasks, they have no assigned tasks.
    //Do whatever you want here!
}

Answer 2

CountDownLatch使用

给定计数进行初始化。通过调用 countDown() 方法，此计数递减。等待此计数达到零的线程可以调用await()方法之一。调用 await() 会阻塞线程，直到计数达到零。

您可以使用CountDownLatch主线程来等待完成所有任务。您可以在主线程调用中将大小CountDownLatch声明为任务CountDownLatch latch = new CountDownLatch(3);数await()等待的方法，并且每个任务完成调用countDown()

public void work(final MyWorkUnit pmyprojectOrg)
{
    final List<MyWorkUnit> allOrgs = new ArrayList<MyWorkUnit>();
    if (pmyprojectOrg != null)
    {
        processData(pmyprojectOrg.getmyprojectOrgId());
    } else { 
        allOrgs.addAll(m_myprojectSvc.findAllWithNonEmptyTokens());
        CountDownLatch latch = new CountDownLatch(allOrgs.size());
        // Cue up threads to execute
        for (final MyWorkUnit myprojectOrg : allOrgs)
        {
            m_threadExecutor.execute(new Thread(new Runnable(){
                @Override
                public void run()
                {
                    System.out.println("started.");
                    processData(myprojectOrg.getmyprojectOrgId());
                   latch.countDown();
                }
            }));
        }  
       //After for loop
       latch.await();      

例：

CountDownLatch latch = new CountDownLatch(3);
Waiter      waiter      = new Waiter(latch);
Decrementer decrementer = new Decrementer(latch);
new Thread(waiter)     .start();
new Thread(decrementer).start();
public class Waiter implements Runnable{
    CountDownLatch latch = null;
    public Waiter(CountDownLatch latch) {
        this.latch = latch;
    }
    public void run() {
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Waiter Released");
    }
}

公共类 Decrementer 实现 Runnable {

CountDownLatch latch = null;
public Decrementer(CountDownLatch latch) {
    this.latch = latch;
}

 public void run() {
        try {
            Thread.sleep(1000);
            this.latch.countDown();
            Thread.sleep(1000);
            this.latch.countDown();
            Thread.sleep(1000);
            this.latch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Answer 3

由于我使用的是 Spring 的 ThreadPoolTaskExecutor，我找到了下面适合我需求的......

protected void waitForThreadPool(final ThreadPoolTaskExecutor threadPoolExecutor)
{
    threadPoolExecutor.setWaitForTasksToCompleteOnShutdown(true);
    threadPoolExecutor.shutdown();    
    try {
        threadPoolExecutor.getThreadPoolExecutor().awaitTermination(30, TimeUnit.SECONDS);
    } catch (IllegalStateException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
}   // waitForThreadPool