原文: https://howtodoinjava.com/java/collections/transferqueue-linkedtransferqueue/
Java TransferQueue
是并发阻塞队列的实现,在这种实现中,生产者可以等待消费者收到消息。LinkedTransferQueue
类是 Java 中TransferQueue
的实现。
例如,TransferQueue
可能在消息传递应用程序中很有用,在该应用程序中,生产者有时(使用方法transfer()
)通过消费者调用take
或poll
来等待元素的接收,而在其他时候则排队元素(通过方法put()
)而无需等待收货。
当生产者到达
TransferQueue
传输消息并且有消费者在等待接收消息时,生产者直接将消息传输给消费者。如果没有消费者等待,那么生产者将不会直接放置消息并返回,而是将等待任何消费者可以使用该消息。
1. LinkedTransferQueue
特性
让我们记下 Java 中LinkedTransferQueue
的几个要点。
LinkedTransferQueue
是链接节点上的无限制队列。- 此队列针对任何给定的生产者对元素 FIFO(先进先出)进行排序。
- 元素插入到尾部,并从队列的开头检索。
- 它提供阻塞插入和检索操作。
- 它不允许使用
NULL
对象。 LinkedTransferQueue
是线程安全的。- 由于异步性质,
size()
方法不是固定时间操作,因此,如果在遍历期间修改此集合,则可能会报告不正确的结果。 - 不保证批量操作
addAll
,removeAll
,retainAll
,containsAll
,equals
和toArray
是原子执行的。 例如,与addAll
操作并发操作的迭代器可能仅查看某些添加的元素。
2. Java LinkedTransferQueue
示例
2.1 LinkedTransferQueue
示例
一个非常简单的示例,用于从LinkedTransferQueue
添加和轮询消息。
LinkedTransferQueue<Integer> linkedTransferQueue = new LinkedTransferQueue<>();
linkedTransferQueue.put(1);
System.out.println("Added Message = 1");
Integer message = linkedTransferQueue.poll();
System.out.println("Recieved Message = " + message);
程序输出。
Added Message = 1
Recieved Message = 1
2.2 LinkedTransferQueue
阻塞插入和检索示例
使用阻塞插入和检索从LinkedTransferQueue
放入和取出元素的 Java 示例。
- 生产者线程将等待,直到有消费者准备从队列中取出项目为止。
- 如果队列为空,使用者线程将等待。 队列中只有一个元素时,它将取出该元素。 只有在消费者接受了消息之后,生产者才可以再发送一条消息。
import java.util.Random;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
public class LinkedTransferQueueExample
{
public static void main(String[] args) throws InterruptedException
{
LinkedTransferQueue<Integer> linkedTransferQueue = new LinkedTransferQueue<>();
new Thread(() ->
{
Random random = new Random(1);
try
{
while (true)
{
System.out.println("Producer is waiting to transfer message...");
Integer message = random.nextInt();
boolean added = linkedTransferQueue.tryTransfer(message);
if(added) {
System.out.println("Producer added the message - " + message);
}
Thread.sleep(TimeUnit.SECONDS.toMillis(3));
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() ->
{
try
{
while (true)
{
System.out.println("Consumer is waiting to take message...");
Integer message = linkedTransferQueue.take();
System.out.println("Consumer recieved the message - " + message);
Thread.sleep(TimeUnit.SECONDS.toMillis(3));
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
程序输出:
Producer is waiting to transfer message...
Consumer is waiting to take message...
Producer is waiting to transfer message...
Producer added the message - 431529176
Consumer recieved the message - 431529176
Consumer is waiting to take message...
Producer is waiting to transfer message...
Producer added the message - 1761283695
Consumer recieved the message - 1761283695
Consumer is waiting to take message...
Producer is waiting to transfer message...
Producer added the message - 1749940626
Consumer recieved the message - 1749940626
Consumer is waiting to take message...
Producer is waiting to transfer message...
Producer added the message - 892128508
Consumer recieved the message - 892128508
Consumer is waiting to take message...
Producer is waiting to transfer message...
Producer added the message - 155629808
Consumer recieved the message - 155629808
请注意,控制台中可能有一些打印语句,似乎消费者甚至在生产者生成消息之前就已经使用了该消息。 不要混淆,这是因为示例的并发性。 实际上,它可以按预期工作。
3. Java LinkedTransferQueue
构造器
LinkedTransferQueue
类提供了 3 种不同的方法来用 Java 构造队列。
LinkedTransferQueue()
:构造一个最初为空的LinkedTransferQueue
。LinkedTransferQueue(Collection c)
:构造一个LinkedTransferQueue
,最初包含给定集合的元素,并以集合迭代器的遍历顺序添加。
4. Java LinkedTransferQueue
方法
您应该知道LinkedTransferQueue
类在下面提供了重要的方法。
Object take()
:检索并删除此队列的头,如有必要,请等待直到元素可用。void transfer(Object o)
:将元素传输到使用者,如有必要,请等待。boolean tryTransfer(Object o)
:如果可能,立即将元素传输到等待的使用者。boolean tryTransfer(Object o, long timeout, TimeUnit unit)
:如果有可能,则在超时之前将元素传送给使用者。int getWaitingConsumerCount()
:返回等待通过BlockingQueue.take()
或定时轮询接收元素的使用者数量的估计值。boolean hasWaitingConsumer()
:如果至少有一个使用者正在等待通过BlockingQueue.take()
或定时轮询接收元素,则返回true
。void put(Object o)
:将指定的元素插入此队列的尾部。boolean add(object)
:将指定的元素插入此队列的末尾。boolean offer(object)
:将指定的元素插入此队列的末尾。boolean remove(object)
:从此队列中移除指定元素的单个实例(如果存在)。Object peek()
:检索但不删除此队列的头,如果此队列为空,则返回 null。Object poll()
:检索并删除此队列的头部,如果此队列为空,则返回 null。Object poll(timeout, timeUnit)
:检索并删除此队列的开头,并在必要时等待指定的等待时间以使元素可用。void clear()
:从此队列中删除所有元素。boolean contains(Object o)
:如果此队列包含指定的元素,则返回true
。Iterator iterator()
:按适当的顺序返回此队列中元素的迭代器。int size()
:返回此队列中的元素数。int drainTo(Collection c)
:从此队列中删除所有可用元素,并将它们添加到给定的集合中。int drainTo(Collection c, int maxElements)
:从此队列中最多移除给定数量的可用元素,并将其添加到给定的集合中。int remainingCapacity()
:返回此队列理想情况下(在没有内存或资源约束的情况下)可以接受而不阻塞的其他元素的数量。Object[] toArray()
:按正确的顺序返回包含此队列中所有元素的数组。
5. Java TransferQueue
总结
在此 Java LinkedTransferQueue
教程中,我们学习了使用LinkedTransferQueue
类,这是一个并发阻塞队列实现,生产者可以在其中等待使用者接收消息。
我们还了解了LinkedTransferQueue
类的一些重要方法和构造器。
将我的问题放在评论部分。
学习愉快!
参考文献: