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

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>

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

    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

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

您可以使用 CountDownLatch 到主线程等待完成所有 task.You 可以在主线程调用 CountDownLatch latch = new CountDownLatch(3); 中声明 CountDownLatch 大小作为任务数 CountDownLatch latch = new CountDownLatch(3); =15=] 方法等待和每个任务完成调用 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");
    }
}

public class 减量器实现 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 2

因为我使用的是 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 3

可以使用ExecutorService创建一个固定的线程池，查看池大小是否为空：

ExecutorService executor = Executors.newFixedThreadPool(50);

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

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

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

How do I wait for my thread factory to finish executing all its tasks?

java

concurrency

spring

multithreading

managedthreadfactory