所有线程完成后，应用程序挂起几分钟(Application hangs for few minutes even after all the threads are done)

 我将一个工作的生产者/消费者示例从Thread / Runnable转换为Executor / Callable / BlockingQueues并使用Poison Pill终止模式。 
 
 如果您运行下面的程序，即使每个线程都已完成，它也会挂起几分钟。 jstack显示在队列上阻塞的大量线程似乎与应用程序无关。 
 
"pool-1-thread-10" prio=5 tid=10b08d000 nid=0x10d91c000 waiting on condition [10d91b000]
   java.lang.Thread.State: TIMED_WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <7f3113510> (a java.util.concurrent.SynchronousQueue$TransferStack)
    at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:198)
    at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:424)
    at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:323)
    at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:874)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:945)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
    at java.lang.Thread.run(Thread.java:680)
 
 我无法弄清楚应用程序挂起的原因。 任何帮助表示赞赏。 谢谢 
 
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducersConsumers {
    private LinkedBlockingQueue<Item> queue = new LinkedBlockingQueue<Item>();
    private static final ExecutorService executorPool = Executors.newCachedThreadPool();
    private Random randGenerator = new Random(System.currentTimeMillis());

    private class Item {
        private boolean done = false;
        private String message;

        private Item(boolean done) {
            this.done = done;
        }

        private Item(String message) {
            this.message = message;
        }

        public boolean isDone() {
            return done;
        }

        public String getMessage() {
            return message;
        }
    }

    private class Producer implements Callable<Long> {
        private final int id;
        private Integer numOfMessages;

        private Producer(int id, int numOfMessages) {
            this.id = id;
            this.numOfMessages = numOfMessages;
        }

        @Override
        public Long call() throws Exception {
            long totalTime = 0;
            while (numOfMessages > 0) {
                String message;
                synchronized (numOfMessages) {
                    long starttime = System.nanoTime();
                    int msgLength = randGenerator.nextInt(20000);
                    StringBuilder sb = new StringBuilder(msgLength);
                    for (int a = 0; a < msgLength; a++) {
                        sb.append((char) ('a' + randGenerator.nextInt(26)));
                    }
                    message = sb.toString();
                    long endtime = System.nanoTime();
                    totalTime += endtime - starttime;
                }
                numOfMessages--;
                queue.put(new Item(message));
            }
            System.out.println("-------------Producer " + id + " is done.");
            queue.put(new Item(true));
            return totalTime;
        }
    }

    private class Consumer implements Callable<Long> {
        private String monitor = "monitor";
        private final int id;

        private Consumer(int id) {
            this.id = id;
        }

        @Override
        public Long call() throws Exception {
            long totalTime = 0;
            while (true) {
                Item item = queue.take();
                if (item.isDone()) {
                    break;
                }
                synchronized (monitor) {
                    long starttime = System.nanoTime();
                    StringBuilder sb = new StringBuilder(item.getMessage());
                    sb = sb.reverse();
                    String message = sb.toString();
                    long endtime = System.nanoTime();
                    totalTime += endtime - starttime;
                }
            }
            System.out.println("+++++++++++++Consumer " + id + " is done.");
            return totalTime;
        }
    }

    public void begin(int threadCount) throws InterruptedException, ExecutionException {
        Collection<Producer> producers = new ArrayList<Producer>();
        for (int i = 0; i < threadCount; i++) {
            producers.add(new Producer(i, randGenerator.nextInt(5)));
        }
        Collection<Consumer> consumers = new ArrayList<Consumer>();
        for (int i = 0; i < threadCount; i++) {
            consumers.add(new Consumer(i));
        }
        try {
            long starttime = System.nanoTime();
            List<Future<Long>> producerFutureList = executorPool.invokeAll(producers);
            List<Future<Long>> consumerFutureList = executorPool.invokeAll(consumers);
            long producerTotalTime = 0;
            long consumerTotalTime = 0;

            for (Future<Long> future : producerFutureList) {
                producerTotalTime += future.get();
            }
            for (Future<Long> future : consumerFutureList) {
                consumerTotalTime += future.get();
            }
            long mainThreadTotalTime = System.nanoTime() - starttime;

            System.out.println("producerTotalTime   " + producerTotalTime);
            System.out.println("consumerTotalTime   " + consumerTotalTime);
            System.out.println("mainThreadTotalTime " + mainThreadTotalTime);
            System.out.println("Difference          " + (producerTotalTime + consumerTotalTime - mainThreadTotalTime));
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            throw e;
        } catch (ExecutionException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            throw e;
        }

    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ProducersConsumers prodcon = new ProducersConsumers();
        prodcon.begin(20);
    }
}

I converted a working Producer/Consumer Example from Thread/Runnable to Executor/Callable/BlockingQueues and using the Poison Pill termination pattern.
 
If you run the program below, it will hang for few minutes even though every thread has completed. jstack shows numerous threads blocked on a queue that is not seemingly related to the application.
 
"pool-1-thread-10" prio=5 tid=10b08d000 nid=0x10d91c000 waiting on condition [10d91b000]
   java.lang.Thread.State: TIMED_WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <7f3113510> (a java.util.concurrent.SynchronousQueue$TransferStack)
    at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:198)
    at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:424)
    at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:323)
    at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:874)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:945)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
    at java.lang.Thread.run(Thread.java:680)
 
I can not figure out why the application hangs. Any help is appreciated. Thank you
 
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducersConsumers {
    private LinkedBlockingQueue<Item> queue = new LinkedBlockingQueue<Item>();
    private static final ExecutorService executorPool = Executors.newCachedThreadPool();
    private Random randGenerator = new Random(System.currentTimeMillis());

    private class Item {
        private boolean done = false;
        private String message;

        private Item(boolean done) {
            this.done = done;
        }

        private Item(String message) {
            this.message = message;
        }

        public boolean isDone() {
            return done;
        }

        public String getMessage() {
            return message;
        }
    }

    private class Producer implements Callable<Long> {
        private final int id;
        private Integer numOfMessages;

        private Producer(int id, int numOfMessages) {
            this.id = id;
            this.numOfMessages = numOfMessages;
        }

        @Override
        public Long call() throws Exception {
            long totalTime = 0;
            while (numOfMessages > 0) {
                String message;
                synchronized (numOfMessages) {
                    long starttime = System.nanoTime();
                    int msgLength = randGenerator.nextInt(20000);
                    StringBuilder sb = new StringBuilder(msgLength);
                    for (int a = 0; a < msgLength; a++) {
                        sb.append((char) ('a' + randGenerator.nextInt(26)));
                    }
                    message = sb.toString();
                    long endtime = System.nanoTime();
                    totalTime += endtime - starttime;
                }
                numOfMessages--;
                queue.put(new Item(message));
            }
            System.out.println("-------------Producer " + id + " is done.");
            queue.put(new Item(true));
            return totalTime;
        }
    }

    private class Consumer implements Callable<Long> {
        private String monitor = "monitor";
        private final int id;

        private Consumer(int id) {
            this.id = id;
        }

        @Override
        public Long call() throws Exception {
            long totalTime = 0;
            while (true) {
                Item item = queue.take();
                if (item.isDone()) {
                    break;
                }
                synchronized (monitor) {
                    long starttime = System.nanoTime();
                    StringBuilder sb = new StringBuilder(item.getMessage());
                    sb = sb.reverse();
                    String message = sb.toString();
                    long endtime = System.nanoTime();
                    totalTime += endtime - starttime;
                }
            }
            System.out.println("+++++++++++++Consumer " + id + " is done.");
            return totalTime;
        }
    }

    public void begin(int threadCount) throws InterruptedException, ExecutionException {
        Collection<Producer> producers = new ArrayList<Producer>();
        for (int i = 0; i < threadCount; i++) {
            producers.add(new Producer(i, randGenerator.nextInt(5)));
        }
        Collection<Consumer> consumers = new ArrayList<Consumer>();
        for (int i = 0; i < threadCount; i++) {
            consumers.add(new Consumer(i));
        }
        try {
            long starttime = System.nanoTime();
            List<Future<Long>> producerFutureList = executorPool.invokeAll(producers);
            List<Future<Long>> consumerFutureList = executorPool.invokeAll(consumers);
            long producerTotalTime = 0;
            long consumerTotalTime = 0;

            for (Future<Long> future : producerFutureList) {
                producerTotalTime += future.get();
            }
            for (Future<Long> future : consumerFutureList) {
                consumerTotalTime += future.get();
            }
            long mainThreadTotalTime = System.nanoTime() - starttime;

            System.out.println("producerTotalTime   " + producerTotalTime);
            System.out.println("consumerTotalTime   " + consumerTotalTime);
            System.out.println("mainThreadTotalTime " + mainThreadTotalTime);
            System.out.println("Difference          " + (producerTotalTime + consumerTotalTime - mainThreadTotalTime));
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            throw e;
        } catch (ExecutionException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            throw e;
        }

    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ProducersConsumers prodcon = new ProducersConsumers();
        prodcon.begin(20);
    }
}

原文：https://stackoverflow.com/questions/6767961