如何等待线程工厂完成其所有任务？

Question

我正在使用Spring 4.3.8。发布Java7。我想创建一个线程工厂来帮助管理应用程序中的某些工作人员。我像这样声明我的线程工厂

<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

由于我使用的是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

Answer 2

使用给定计数初始化CountDownLatch。调用countDown（）方法可减少此计数。等待此计数达到零的线程可以调用其中一个方法。调用wait（）会阻塞线程，直到计数为零。

您可以使用CountDownLatch到主线程等待完成所有任务。您可以声明CountDownLatch大小为任务数CountDownLatch latch=new CountDownLatch（3）；在主线程调用wait（）方法等待和每个任务完成调用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

您可以使用ExecutorService创建固定线程池并检查池大小是否为空：

ExecutorService executor = Executors.newFixedThreadPool(50);

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

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!
}