logo标签列表

Executors.newFixedThreadPool无法终止运行

javamultithreadingthreadpool
3

我正在使用ConcurrentLinkedQueue来存储计算步骤，并使用由Executors.newFixedThreadPool创建的ExecutorService来执行它们。我的问题是应用程序从未终止。以下是一些代码：

public class Run {

    public static void main (String[] args) throws Exception {
        ParallelExecutor executor = new ParallelExecutor();
        executor.execute();

        // manual shutdown
        Thread.sleep(30 * 1000);
        executor.stop();

        // the main thread dies, the application keeps running
}

 

public class ParallelExecutor implements Runnable {

    // stores executions steps to
    private ConcurrentLinkedQueue<ExecutionStep> queue = new ConcurrentLinkedQueue<ExecutionStep>();

    private ExecutorService threadPoolExecutor = Executors.newFixedThreadPool(10);

    private Thread feedingThread = new Thread(this);

    public void execute() {
        ExecutionStep step = new ConcreteExecutionStep();
        this.queue.add(step);
        this.feedingThread.start();
    }

    public void stop() {
        if (log.isInfoEnabled()) {
            log.info("Shutting down");
        }
        this.queue = new ConcurrentLinkedQueue<ExecutionStep>();

        try {
            this.feedingThread.join(0);
            this.threadPoolExecutor.shutdownNow();
        } catch (InterruptedException e) {
            log.warn("Something happened!", e);
        }
    }

    public void run() {
        while (true) {

            if (this.threadPoolExecutor.isShutdown()) {
                return;
            }

            try {
                Thread.sleep(this.waitMillis);
            } catch (InterruptedException e) {
                // NOP
            }

            if (!this.queue.isEmpty()) {
                this.threadPoolExecutor.execute(new Runnable() {

                    @Override
                    public void run() {
                        ExecutionStep step = this.queue.poll();
                        List<ExecutionStep> nextSteps = Collections.emptyList();
                        try {
                            nextSteps = step.execute();
                        } catch (Exception e) {
                            // NOP
                        }

                        // here we feed the queue
                        this.queue.addAll(nextSteps);
                    }
                });
            }
        }
    }

编辑: 主线程死后这里是线程转储:

2011-09-06 17:20:56
Full thread dump Java HotSpot(TM) 64-Bit Server VM (20.1-b02-384 mixed mode):

"DestroyJavaVM" prio=5 tid=10f874800 nid=0x100501000 waiting on condition [00000000]
   java.lang.Thread.State: RUNNABLE

"Poller SunPKCS11-Darwin" daemon prio=1 tid=101a4d800 nid=0x11362e000 waiting on condition [11362d000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
    at java.lang.Thread.sleep(Native Method)
    at sun.security.pkcs11.SunPKCS11$TokenPoller.run(SunPKCS11.java:692)
    at java.lang.Thread.run(Thread.java:680)

"pool-2-thread-1" prio=5 tid=103db3000 nid=0x11341a000 waiting on condition [113419000]
   java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <7f42e8f38> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1987)
    at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:399)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
    at java.lang.Thread.run(Thread.java:680)

"Timer-0" daemon prio=5 tid=101a63000 nid=0x113317000 in Object.wait() [113316000]
   java.lang.Thread.State: TIMED_WAITING (on object monitor)
    at java.lang.Object.wait(Native Method)
    - waiting on <7f49cc200> (a java.util.TaskQueue)
    at java.util.TimerThread.mainLoop(Timer.java:509)
    - locked <7f49cc200> (a java.util.TaskQueue)
    at java.util.TimerThread.run(Timer.java:462)

"Low Memory Detector" daemon prio=5 tid=10180a000 nid=0x10f607000 runnable [00000000]
   java.lang.Thread.State: RUNNABLE

"C2 CompilerThread1" daemon prio=9 tid=101809800 nid=0x10f504000 waiting on condition [00000000]
   java.lang.Thread.State: RUNNABLE

"C2 CompilerThread0" daemon prio=9 tid=103951000 nid=0x10f401000 waiting on condition [00000000]
   java.lang.Thread.State: RUNNABLE

"JDWP Command Reader" daemon prio=5 tid=103950000 nid=0x10df01000 runnable [00000000]
   java.lang.Thread.State: RUNNABLE

"JDWP Event Helper Thread" daemon prio=5 tid=10394f800 nid=0x10dc0a000 runnable [00000000]
   java.lang.Thread.State: RUNNABLE

"JDWP Transport Listener: dt_socket" daemon prio=5 tid=10394e800 nid=0x10db07000 runnable [00000000]
   java.lang.Thread.State: RUNNABLE

"Signal Dispatcher" daemon prio=9 tid=10e000000 nid=0x10da04000 waiting on condition [00000000]
   java.lang.Thread.State: RUNNABLE

"Surrogate Locker Thread (Concurrent GC)" daemon prio=5 tid=101808800 nid=0x10d901000 waiting on condition [00000000]
   java.lang.Thread.State: RUNNABLE

"Finalizer" daemon prio=8 tid=10393f000 nid=0x10c204000 in Object.wait() [10c203000]
   java.lang.Thread.State: WAITING (on object monitor)
    at java.lang.Object.wait(Native Method)
    - waiting on <7f44e3d70> (a java.lang.ref.ReferenceQueue$Lock)
    at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118)
    - locked <7f44e3d70> (a java.lang.ref.ReferenceQueue$Lock)
    at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134)
    at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:159)

"Reference Handler" daemon prio=10 tid=10393e000 nid=0x10c101000 in Object.wait() [10c100000]
   java.lang.Thread.State: WAITING (on object monitor)
    at java.lang.Object.wait(Native Method)
    - waiting on <7f44e5220> (a java.lang.ref.Reference$Lock)
    at java.lang.Object.wait(Object.java:485)
    at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:116)
    - locked <7f44e5220> (a java.lang.ref.Reference$Lock)

"VM Thread" prio=9 tid=103939800 nid=0x10b6fc000 runnable 

"Gang worker#0 (Parallel GC Threads)" prio=9 tid=103801800 nid=0x102601000 runnable 

"Gang worker#1 (Parallel GC Threads)" prio=9 tid=103802000 nid=0x102704000 runnable 

"Gang worker#2 (Parallel GC Threads)" prio=9 tid=103803000 nid=0x107102000 runnable 

"Gang worker#3 (Parallel GC Threads)" prio=9 tid=103803800 nid=0x107205000 runnable 

"Gang worker#4 (Parallel GC Threads)" prio=9 tid=103804000 nid=0x107308000 runnable 

"Gang worker#5 (Parallel GC Threads)" prio=9 tid=103804800 nid=0x10740b000 runnable 

"Gang worker#6 (Parallel GC Threads)" prio=9 tid=103805800 nid=0x10750e000 runnable 

"Gang worker#7 (Parallel GC Threads)" prio=9 tid=103806000 nid=0x107611000 runnable 

"Concurrent Mark-Sweep GC Thread" prio=9 tid=1038e3800 nid=0x10b408000 runnable 
"Gang worker#0 (Parallel CMS Threads)" prio=9 tid=1038e2000 nid=0x10aa02000 runnable 

"Gang worker#1 (Parallel CMS Threads)" prio=9 tid=1038e2800 nid=0x10ab05000 runnable 

"VM Periodic Task Thread" prio=10 tid=10181c000 nid=0x10f70a000 waiting on condition 

"Exception Catcher Thread" prio=10 tid=103801000 nid=0x1017f9000 runnable 
JNI global references: 13614

Heap
 par new generation   total 19136K, used 16017K [7f3000000, 7f44c0000, 7f44c0000)
  eden space 17024K,  81% used [7f3000000, 7f3d944a8, 7f40a0000)
  from space 2112K, 100% used [7f42b0000, 7f44c0000, 7f44c0000)
  to   space 2112K,   0% used [7f40a0000, 7f40a0000, 7f42b0000)
 concurrent mark-sweep generation total 63872K, used 6053K [7f44c0000, 7f8320000, 7fae00000)
 concurrent-mark-sweep perm gen total 30656K, used 30476K [7fae00000, 7fcbf0000, 800000000)

编辑：问题是由于在我发布问题之前剪切的代码中启动了另一个执行器（静默启动），因为我认为它与问题无关。无论如何，Peter Lawrey 的答案仍然是有价值的参考资料，因此被接受。

如果你有一个线程池和队列的ExecutorService，为什么还需要另一个线程和另一个队列来添加任务呢？我会让执行器池完成所有操作，这样可以简化关闭它的过程。如果没有额外的线程和队列，代码会更简单。 - Peter Lawrey
nextSteps = step.execute(); ：这个类（未显示）是否没有任何问题？ - Mister Smith
3个回答
3

在关闭执行器之前，您正在等待feedingThread.join(0)终止，而饲料线程将继续在其while(true)-loop中运行，直到执行器停止。

但是这种行为与他在主函数结尾处的评论“主线程死亡”相矛盾。那么他说错了吗？ - akappa
我确认主线程已经停止。我编辑了我的原始问题，以显示在每次执行结束时无休止等待状态的线程转储。 - skuro
我也尝试在关闭线程执行器之后加入feedingThread，但行为相同。 - skuro
2

只要没有工作正在进行，它就会关闭，并且不会不时暂停，这会减慢您的应用程序并使其看起来好像什么都没有发生。

public class ParallelExecutor {
  interface ExecutionStep {
    List<ExecutionStep> execute();
  }

  public static final int N_THREADS = Runtime.getRuntime().availableProcessors();
  private final ExecutorService executorService = Executors.newFixedThreadPool(N_THREADS);

  public void execute(final ExecutionStep step) {
    executorService.submit(new Runnable() {
      @Override
      public void run() {
        for (ExecutionStep es : step.execute())
          execute(es);
      }
    });
  }

  public void stop() {
    this.executorService.shutdownNow();
  }
}

运行

ParallelExecutor executor = new ParallelExecutor();
executor.execute(new ConcreteExecutionStep());

// manual shutdown
Thread.sleep(30 * 1000);
executor.stop();
1
去除不必要的队列肯定是有意义的。我会重构我的代码，测试你的建议，并告诉你结果。谢谢。 - skuro
2
感谢您的输入，问题最终被证明是由于这里的某些看不见的东西引起的，因为我删除了相关代码，认为它是无关紧要的 :-) 无论如何，您的答案对于每个在SO上搜索执行者提示的人仍然非常有用。 - skuro
0

我找到了一个非常简单的答案。如果你不仅从主函数中返回，而是以 System.exit(0) 结束它，执行器将按预期终止。

4
你也可以拔掉电脑的插头。 - Ted Bigham
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接