java - BlockingQueue 和释放线程等待

Question

假设我有 BlockingQueue 并且一些线程被称为 take() 但此时队列是空的。假设我以某种方式知道将来不会有新元素出现在队列中。如何释放那些被称为 take() 的线程等待？谢谢!

  public void run() {
        //noinspection InfiniteLoopStatement
        while (true) {
            try {
                Thread.sleep(millisecondsToSleep);
                if (!kitchen.processedOrdersEmpty()) {
                    Order processedOrder = kitchen.getFromProcessedOrders();
                    kitchen.printMessage("took order#" + processedOrder.getOrderNumber());
                    Thread.sleep(millisecondsToServe);
                    kitchen.printMessage("served order#" + processedOrder.getOrderNumber());
                } else {
                    int currentRandom = getNextRandom();
                    if (currentRandom <= 10) {
                        Order newOrder = new Order(kitchen.getLastOrderNumber());
                        kitchen.puIntoUnprocessedOrders(newOrder);
                        kitchen.printMessage("generated new order#" + newOrder.getOrderNumber());
                    } else {
                        Thread.sleep(millisecondsToSleep);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

kitchen.getFromProcessedOrders() 等于:

public Order getFromProcessedOrders() throws InterruptedException {
    return processedOrders.take();
} 

processedOrders 是一个 BlockingQueue

Answer 1

根据 java docs -

A BlockingQueue does not intrinsically support any kind of "close" or "shutdown" operation to indicate that no more items will be added. The needs and usage of such features tend to be implementation-dependent. For example, a common tactic is for producers to insert special end-of-stream or poison objects, that are interpreted accordingly when taken by consumers.

你应该在它被阻塞的地方中断线程 -

public void run() {
        //noinspection InfiniteLoopStatement
        while (true) {
            try {
                Thread.sleep(millisecondsToSleep);
                if (!kitchen.processedOrdersEmpty()) {
                    Order processedOrder = kitchen.getFromProcessedOrders();
                    kitchen.printMessage("took order#" + processedOrder.getOrderNumber());
                    Thread.sleep(millisecondsToServe);
                    kitchen.printMessage("served order#" + processedOrder.getOrderNumber());
                } else {
                    int currentRandom = getNextRandom();
                    if (currentRandom <= 10) {
                        Order newOrder = new Order(kitchen.getLastOrderNumber());
                        kitchen.puIntoUnprocessedOrders(newOrder);
                        kitchen.printMessage("generated new order#" + newOrder.getOrderNumber());
                    } else {
                        Thread.sleep(millisecondsToSleep);
                    }
                }
            } catch (InterruptedException ex) { 
                //... handle ...
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    }