Java中的 LinkedTransferQueue getWaitingConsumerCount() 方法及示例

Java.util.concurrent.LinkedTransferQueue.getWaitingConsumerCount()方法返回等待通过 take() 或定时轮询从 LinkedTransferQueue(LTQ) 接收元素的消费者数量。这里的消费者将从 LTQ 中读取元素。生产者将向 LTQ 添加元素。该方法仅在多个线程运行时有用，即生产者正在传输消息而消费者同时接收它们。消费者将尝试读取 LTQ 的头部。如果他们不能（空 LTQ 或 head 已经被占用），则称他们正在等待。消费者将等到他们开始阅读头部，或者直到某个时间限制过去。当生产者通过 transfer() 添加到 LTQ 时，就说它被阻塞了。这意味着生产者必须等到消费者读取元素才能将另一个元素添加到 LTQ。

句法：

public int getWaitingConsumerCount()

参数：此方法不接受任何参数。

返回值：该方法返回一个int值，即等待LTQ头部的消费者数量，通过take()获取。

示例：下面的程序使用 1 个生产者和 3 个消费者。生产者发送 3 条消息。首先调用 take() 的消费者将收到最新消息。因此，该程序的不同执行会导致不同的消费者获得不同的消息。

// Java program to show LinkedTransferQueue
// getWaitingConsumerCount() method
  
import java.util.concurrent.*;
  
public class GFG {
  
    // create object of LinkedTransferQueue
    LinkedTransferQueue LTQ
        = new LinkedTransferQueue<>();
  
    class Producer implements Runnable {
  
        @Override
        public void run()
        {
            for (int i = 0; i < 3; i++) {
                try {
                    System.out.println("Producer is"
                                       + " waiting to transfer...");
  
                    // add 0, 1, 2 to LTQ
                    LTQ.transfer(i);
                    System.out.println("Producer "
                                       + "transferred element: " + i);
  
                    // display how many consumers are waiting
                    // after producer has transferred element
                    System.out.println("Waiting consumer "
                                       + "count after transfer: "
                                       + LTQ
                                             .getWaitingConsumerCount());
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
  
    class Consumer implements Runnable {
        int id;
        Consumer(int id)
        {
            this.id = id;
        }
  
        @Override
        public void run()
        {
            try {
                System.out.println("Consumer "
                                   + id
                                   + " is waiting to "
                                   + "take element...");
  
                // display how many consumers are waiting
                // before a consumer has taken element from LTQ
                System.out.println("Waiting consumer"
                                   + " count before take: "
                                   + LTQ
                                         .getWaitingConsumerCount());
  
                Integer s = LTQ.take();
                System.out.println("Consumer "
                                   + id
                                   + " received Element: "
                                   + s);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
  
    public static void main(String[] args) throws InterruptedException
    {
        // class object to keep common LTQ for all consumers
        //(alternative: LTQ can be declared static)
        GFG obj = new GFG();
  
        // enabling multiple threads
        ExecutorService exService
            = Executors.newFixedThreadPool(3);
  
        // declare one producer and 3 consumers
        Producer producer = obj.new Producer();
        Consumer consumer0 = obj.new Consumer(0);
        Consumer consumer1 = obj.new Consumer(1);
        Consumer consumer2 = obj.new Consumer(2);
  
        // start producer and consumers
        exService.execute(producer);
        exService.execute(consumer0);
        exService.execute(consumer1);
        exService.execute(consumer2);
  
        // stop
        exService.shutdown();
    }
}

输出：

注意：显示的是 NetBeans 8.2 中的输出。建议在本地试用此示例，因为它使用了多线程。在线 IDE 服务器可能不支持此功能或让您添加此类负载，从而导致超出时间限制错误。

参考： https://docs.oracle.com/javase/9/docs/api/ Java/util/concurrent/LinkedTransferQueue.html#getWaitingConsumerCount–