TransferQueue是Java并发包中的一个强大工具，专为需要精确的数据传递场景而设计。它实现了BlockingQueue接口，但提供了更为独特的transfer和tryTransfer方法，允许生产者和消费者之间以同步的方式进行数据的直接传递。当生产者生成数据项时，它可以等待直到有消费者准备好接收；同样地，当消费者需要数据时，它可以等待直到有生产者提供。这种机制确保了数据在传递过程中的一致性和准确性，是构建高性能、低延迟并发系统的关键组件之一。
目录
- 1️⃣概述
- 2️⃣BlockingQueue vs TransferQueue
- 3️⃣核心方法
- 4️⃣常见实现
- 5️⃣使用场景
- 6️⃣LinkedTransferQueue实现生产者-消费者场景
- 7️⃣总结
1️⃣概述
TransferQueue是Java并发包java.util.concurrent中的一个接口，它扩展了BlockingQueue接口。与传统的BlockingQueue不同，TransferQueue提供了更精确的控制，允许生产者和消费者线程之间进行更直接的交互。它引入了两种新的操作方法：transfer(E e)和tryTransfer(E e, long timeout, TimeUnit unit)，这两种方法提供了在数据可用时的等待/传输语义。
2️⃣BlockingQueue vs TransferQueue
在深入了解TransferQueue之前，让我们先回顾一下BlockingQueue。BlockingQueue是一个线程安全的队列，它支持在尝试检索元素但队列为空时等待，以及尝试添加元素但队列已满时等待。它是实现生产者-消费者模式的一种常见方式。
然而，在某些情况下，您可能需要更细粒度的控制，以确保当一个线程正在等待接收数据时，有一个对应的线程准备发送数据。这就是TransferQueue派上用场的地方。与BlockingQueue不同，TransferQueue的实现会尝试立即满足一个take或put操作的要求，如果不能立即满足，那么等待的线程将会被“匹配”到一个即将进入的相反操作。
3️⃣核心方法
TransferQueue接口声明了以下关键方法：
E transfer(E e)- 将元素传输给消费者，如果可能的话，否则等待直到一个消费者准备接收它。boolean tryTransfer(E e, long timeout, TimeUnit unit)- 尝试将元素传输给等待的消费者，在指定的时间内等待，如果在给定的时间内无法完成传输，则返回false。E tryTransfer(E e)- 尝试立即将元素传输给等待的消费者，如果不能立即传输，则返回null。E getWaitingConsumer()- 返回正在等待接收元素的线程（如果存在的话），主要是为了监视和调试的目的。int getWaitingProducerCount()- 返回正在等待向此队列传输元素的线程数量。int getWaitingConsumerCount()- 返回正在等待从此队列接收元素的线程数量。
注意，并非所有TransferQueue实现都需要提供所有这些方法的完整实现。某些实现可能不支持全部的操作集，例如tryTransfer的超时版本。
4️⃣常见实现
LinkedTransferQueue是TransferQueue接口的一个常用实现。它是一个基于链表的、无界的阻塞队列。与ArrayBlockingQueue和LinkedBlockingQueue相比，LinkedTransferQueue的传输操作具有不同的特性。
- 公平性：与一些其他阻塞队列不同，
LinkedTransferQueue通常遵循FIFO原则，但不保证元素的顺序在多生产者和多消费者环境下绝对精确。 - 无界：
LinkedTransferQueue在逻辑上是无界的，这意味着你可以放入任意多的元素，只要你的程序有足够的内存来处理它们。然而，在实践中，队列的容量受到JVM可用内存的限制。 - 高效的传输操作：
LinkedTransferQueue使用一种称为"双重数据链接"的策略，使得传输操作可以在恒定的时间内完成，而与队列中元素的数量无关。
5️⃣使用场景
TransferQueue通常用于以下场景：
- 当需要在生产者线程和消费者线程之间进行精确匹配时，以确保生产者的数据可以立即被消费者处理。
- 当生产者需要等待消费者准备好接收数据，而不仅仅是等待空间在队列中变得可用时。
- 当你想要利用Java并发包的强大功能来实现高级的多线程协调策略时。
6️⃣LinkedTransferQueue实现生产者-消费者场景
下面代码使用LinkedTransferQueue实现一个简单的生产者-消费者场景，其中生产者生成数据并将其传输给消费者，消费者处理这些数据。
import java.util.concurrent.*;
public class TransferQueueDemo {
// 定义生产的数据类型
static class DataItem {
int id;
public DataItem(int id) {
this.id = id;
}
@Override
public String toString() {
return "DataItem{" + "id=" + id + '}';
}
}
// 生产者任务
static class Producer implements Runnable {
private final TransferQueue<DataItem> queue;
public Producer(TransferQueue<DataItem> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 0; i < 10; i++) {
DataItem item = new DataItem(i);
System.out.println("生产者生产了数据: " + item);
// 将数据传输给消费者，如果消费者未准备好，生产者将等待
queue.transfer(item);
// 模拟生产耗时
Thread.sleep(1000);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
// 消费者任务
static class Consumer implements Runnable {
private final TransferQueue<DataItem> queue;
public Consumer(TransferQueue<DataItem> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
// 从队列中接收数据，如果没有数据，消费者将等待
DataItem item = queue.take();
System.out.println("消费者消费了数据: " + item);
// 模拟消费耗时
Thread.sleep(1500);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
public static void main(String[] args) {
// 创建一个LinkedTransferQueue实例
TransferQueue<DataItem> queue = new LinkedTransferQueue<>();
// 启动生产者线程
Thread producerThread = new Thread(new Producer(queue));
producerThread.start();
// 启动消费者线程
Thread consumerThread = new Thread(new Consumer(queue));
consumerThread.start();
}
}
-
我们定义了一个名为
DataItem的简单类来持有数据。Producer类实现了Runnable接口，并在其run方法中循环生成DataItem对象，并使用transfer方法将它们放入TransferQueue。如果此时没有消费者在等待接收数据，生产者线程将会阻塞。 -
Consumer类也实现了Runnable接口，并在其run方法中无限循环地从TransferQueue中取出数据项（通过take方法），然后模拟消费这些数据。如果队列为空，消费者线程将会阻塞，直到生产者放入新的数据项。 -
在
main方法中，我们创建了一个LinkedTransferQueue实例，并分别启动了一个生产者和一个消费者线程。这个程序将持续运行，直到被外部中断或者手动停止。
7️⃣总结
TransferQueue是一个功能强大的并发工具，它扩展了标准的阻塞队列概念，允许生产者和消费者之间进行更直接和精确的数据传输。通过使用TransferQueue，你可以构建更复杂、更高效的多线程应用程序，同时减少因资源竞争而导致的线程等待时间。在选择TransferQueue时，请考虑你的应用程序是否需要这种高级别的控制和协调，以及你选择的TransferQueue实现是否满足你的性能和功能需求。
