SynchronousQueue 源码解析
引言
在 Java 并发编程领域，SynchronousQueue作为 Java 并发包中的一员，占据着独特而关键的地位，发挥着不可替代的重要作用。它是一种特殊的阻塞队列，与常见的有界队列（如ArrayBlockingQueue）和无界队列（如LinkedBlockingQueue）有着显著区别，其最突出的特点是没有容量，这意味着它无法像其他队列那样存储元素 。在SynchronousQueue中，每一个插入操作都必须等待一个对应的删除操作，反之亦然，这种直接的线程间协作机制，使得它成为了线程间高效通信和数据传递的利器。
以一个简单的生产 - 消费场景为例，假设有一个订单处理系统，生产者线程负责生成订单，消费者线程负责处理订单。在使用SynchronousQueue时，当生产者线程生成一个订单并尝试将其放入队列时，如果此时没有消费者线程准备好接收订单，生产者线程将会被阻塞，直到有消费者线程来取走订单；反之，消费者线程在尝试从队列中获取订单时，如果没有生产者线程放入订单，消费者线程也会被阻塞。这种严格的同步机制确保了订单的即时处理，避免了订单在队列中的积压，大大提高了系统的响应速度和处理效率，特别适用于对数据传输实时性要求极高的场景。
SynchronousQueue在 Java 线程池的实现中也扮演着重要角色，如Executors.newCachedThreadPool()方法就使用了SynchronousQueue。在这种线程池配置下，任务被提交后会立即被执行，如果没有可用线程，任务会等待线程变为可用，而不是像其他队列那样将任务缓存起来，从而实现了高效的任务调度。
SynchronousQueue 基础认知
定义与特点
SynchronousQueue是 Java 并发包中一个独特的存在，它被定义为一个没有容量的阻塞队列。从本质上来说，它更像是一个线程间直接移交元素的通道，而非传统意义上存储元素的队列 。这意味着，在SynchronousQueue中，每一个插入操作（put）都必须等待另一个线程执行对应的移除操作（take），反之亦然，这种特性使得它在某些特定场景下能够实现高效的线程间协作。
为了更直观地理解SynchronousQueue的特点，我们来看一段简单的代码示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class SynchronousQueueDemo {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName() + " put 2");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName() + " put 3");
blockingQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "T1").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "T2").start();
}
}在这段代码中，我们创建了一个SynchronousQueue实例，并启动了两个线程：T1和T2。T1线程尝试向队列中放入三个元素，T2线程在延迟 3 秒后开始从队列中取出元素。由于SynchronousQueue没有容量，T1线程在执行put("1")后，会被阻塞，直到T2线程执行take()操作将元素取出。当T2线程取出第一个元素后，T1线程才会继续执行下一个put操作，以此类推。通过这个例子，可以清晰地看到SynchronousQueue插入和移除操作必须配对的特性，以及它是如何实现线程间的同步协作的。
此外，SynchronousQueue还具有公平性可选的特点。在创建SynchronousQueue时，可以通过构造函数传入一个布尔值来指定是否采用公平模式。如果采用公平模式，SynchronousQueue会采用公平锁，并配合一个先进先出（FIFO）的队列来管理等待的线程，确保线程按照等待的先后顺序进行操作；如果采用非公平模式（默认），SynchronousQueue会使用后进先出（LIFO）的栈结构来管理等待线程，这种模式在高并发情况下可能会导致某些线程长时间等待，即所谓的 “饥饿” 现象，但在某些场景下，非公平模式由于减少了线程切换的开销，能够提供更高的吞吐量 。
与其他队列的差异
与 Java 中常见的队列，如ArrayBlockingQueue和LinkedBlockingQueue相比，SynchronousQueue有着诸多显著的差异，这些差异使得它在不同的应用场景中发挥着独特的作用。
首先，从容量特性来看，ArrayBlockingQueue是有界队列，在创建时需要指定一个固定的容量，用于存储元素；LinkedBlockingQueue默认是无界队列（也可以指定容量变为有界队列），理论上可以存储无限个元素；而SynchronousQueue则是没有容量，它不存储任何元素，只是作为线程间传递元素的桥梁。例如，当使用ArrayBlockingQueue时，如果队列已满，生产者线程调用put方法会被阻塞，直到有消费者线程从队列中取出元素，腾出空间；LinkedBlockingQueue在无界模式下，生产者线程的put操作不会因为队列满而阻塞；但在SynchronousQueue中，生产者线程的put操作会立即被阻塞，除非有消费者线程正在等待接收元素 。
其次，在操作特性方面，ArrayBlockingQueue和LinkedBlockingQueue都支持异步操作，即生产者和消费者可以在不同的时间独立地进行插入和移除操作。在ArrayBlockingQueue中，生产者线程可以在队列未满时将元素插入队列，消费者线程可以在队列不为空时从队列中取出元素，两者之间不需要严格的配对；LinkedBlockingQueue同样如此，它通过分离的锁机制实现了生产者和消费者操作的并行性 。而SynchronousQueue的操作是严格同步的，生产者线程的插入操作必须等待消费者线程的移除操作，反之亦然，这种严格的同步机制确保了元素的即时传递，但也限制了其在某些场景下的使用。
再者，从数据结构和实现原理上看，ArrayBlockingQueue基于数组实现，内部维护一个定长的数组来存储元素，通过ReentrantLock来保证线程安全；LinkedBlockingQueue基于链表实现，使用单向链表来存储元素，同样通过锁机制来控制并发访问，并且为生产者和消费者分别提供了独立的锁，以提高并发性能；SynchronousQueue则采用了一种更为复杂的机制，它内部有两种实现模式：公平模式下使用TransferQueue（基于链表实现的先进先出队列）来管理等待线程，非公平模式下使用TransferStack（基于栈结构）来管理等待线程，并且通过比较并交换（CAS）操作来实现高效的并发控制，避免了传统锁机制带来的开销 。
另外，在一些方法的行为上，SynchronousQueue也与其他队列不同。例如，SynchronousQueue的isEmpty()方法始终返回true，因为它没有存储元素；remainingCapacity()方法始终返回0；peek()方法始终返回null，因为队列中没有可查看的元素；remove()和removeAll()方法始终返回false，因为队列中没有元素可供删除。而ArrayBlockingQueue和LinkedBlockingQueue的这些方法会根据队列的实际情况返回相应的结果 。
综上所述，SynchronousQueue与其他常见队列在容量、操作特性、数据结构和方法行为等方面都存在明显差异。在实际应用中，开发者需要根据具体的业务需求和场景特点，选择合适的队列来实现高效的并发编程。如果需要即时的数据传递和严格的线程同步，SynchronousQueue是一个不错的选择；如果需要缓存数据以应对生产和消费速度不一致的情况，ArrayBlockingQueue或LinkedBlockingQueue则更为合适 。
SynchronousQueue 架构与数据结构
整体架构概览
SynchronousQueue的内部架构基于Transferer接口，这一接口定义了数据转移的核心操作，是SynchronousQueue实现线程间同步数据传递的关键所在 。Transferer接口只有一个抽象方法transfer，该方法承担着将数据从生产者线程转移到消费者线程（或者反之）的重要职责，它的实现决定了SynchronousQueue的具体行为和性能表现 。
Transferer接口有两个具体的实现类：TransferStack和TransferQueue，分别对应非公平模式和公平模式 。在创建SynchronousQueue实例时，可以通过构造函数的参数fair来指定使用哪种模式 。如果fair为true，则使用TransferQueue实现公平模式，在这种模式下，线程按照等待的先后顺序进行数据传递，遵循先进先出（FIFO）的原则，保证了每个线程都有公平的机会进行操作，避免了线程饥饿问题 ；如果fair为false（默认值），则使用TransferStack实现非公平模式，该模式采用后进先出（LIFO）的栈结构来管理等待线程，在高并发场景下，虽然可能会导致某些线程长时间等待，但由于减少了线程切换的开销，能够在一定程度上提高系统的吞吐量 。
SynchronousQueue的put方法和take方法在本质上都是通过调用Transferer接口的transfer方法来实现的 。当一个线程调用put方法时，它会将数据作为参数传递给transfer方法，尝试将数据转移给等待的消费者线程；如果此时没有消费者线程等待，该线程会被阻塞，直到有消费者线程调用take方法并成功接收数据 。同样，当一个线程调用take方法时，它会向transfer方法传递null作为参数，表示请求获取数据，若当前没有生产者线程提供数据，该线程也会被阻塞，直到有生产者线程调用put方法并将数据转移过来 。这种通过transfer方法实现的统一操作，使得SynchronousQueue的插入和移除操作紧密关联，实现了高效的线程间同步 。
数据结构详解
TransferStack（非公平模式）
TransferStack是SynchronousQueue在非公平模式下的核心数据结构，它基于栈的结构来实现线程间的数据传递和同步，通过独特的节点设计和操作逻辑，实现了高效的并发控制 。
TransferStack中的节点由SNode类表示，每个SNode节点包含了丰富的信息，以支持栈的操作和线程间的协作 。其主要属性包括：
next：一个volatile修饰的SNode类型变量，指向栈中的下一个节点，用于构建栈的链表结构，通过CAS（比较并交换）操作来保证在多线程环境下对next指针的修改是线程安全的 。match：也是volatile修饰的SNode类型变量，用于记录与当前节点匹配的节点 。当一个生产者节点和一个消费者节点成功匹配时，它们的match属性会相互指向对方，从而实现数据的传递 。waiter：volatile修饰的Thread类型变量，保存了当前节点对应的等待线程 。如果当前节点在等待匹配时，对应的线程会被阻塞，通过waiter可以在匹配成功时唤醒该线程 。item：Object类型变量，用于存储节点的数据 。对于生产者节点，item存储要传递的数据；对于消费者节点，item为null，表示请求获取数据 。mode：int类型变量，用于标识节点的类型 。它有三个取值：REQUEST（0）表示该节点是一个消费者请求节点，即等待获取数据的节点；DATA（1）表示该节点是一个生产者数据节点，即包含要传递数据的节点；FULFILLING（2）表示该节点正在参与匹配过程，用于标记正在进行数据传递的节点 。
TransferStack的核心操作是transfer方法，它实现了数据的转移和线程的同步，其主要流程如下：
初始化与模式判断：首先，根据传入的参数
e判断当前操作的模式 。如果e为null，则表示当前是一个消费者线程，操作模式为REQUEST；如果e不为null，则表示当前是一个生产者线程，操作模式为DATA。同时，初始化一个SNode节点s，用于表示当前操作的线程 。自旋操作：进入一个无限循环（自旋），在每次循环中，获取当前栈顶节点
h。栈为空或模式相同处理：如果栈为空（
h == null），或者栈顶节点的模式与当前操作模式相同（即都是REQUEST或都是DATA），则将当前节点s压入栈顶 。通过casHead方法使用CAS操作将head指针更新为当前节点s，如果CAS操作成功，说明压栈成功，然后阻塞当前线程，等待匹配 。当前线程会通过LockSupport.park方法被挂起，直到被其他线程唤醒 。模式互补处理：如果栈顶节点的模式与当前操作模式互补（即一个是
REQUEST，另一个是DATA），则表示找到了匹配的节点 。此时，将当前节点s的模式设置为FULFILLING，并尝试将其压入栈顶 。如果压栈成功，说明匹配成功，然后将栈顶节点（即与当前节点匹配的节点）和当前节点s从栈中弹出 。具体操作是通过casHead方法将head指针更新为栈顶节点的下一个节点，从而实现节点的弹出 。最后，返回匹配节点中的数据（对于生产者线程，返回null；对于消费者线程，返回匹配节点的item） 。帮助匹配处理：如果栈顶节点已经是一个
FULFILLING节点，说明其他线程正在进行匹配操作，当前线程可以帮助其完成匹配 。具体做法是找到与栈顶节点匹配的节点，将它们从栈中弹出，并唤醒匹配节点对应的等待线程 。然后，继续自旋，尝试进行自己的操作 。
E transfer(E e, boolean timed, long nanos) {
SNode s = null; // constructed/reused as needed
// 判断是生产者还是消费者
int mode = (e == null) ? REQUEST : DATA;
for (;;) {
SNode h = head;
// 队列为空，或模式相同
if (h == null || h.mode == mode) { // empty or same-mode
// 不能等待且已超时
if (timed && nanos <= 0) { // can't wait
if (h != null && h.isCancelled())
casHead(h, h.next); // pop cancelled node
else
return null;
// 创建新节点并添加到队列头部
// snode(s, e, h, mode) 方法创建一个新的 SNode 节点
// e：节点携带的数据（任务或结果）,h：当前的头节点，新节点会指向它
// mode：节点模式（DATA、REQUEST 或 FULFILLING）
// casHead(h, s) 尝试原子性地将队列头从 h 更新为 s
} else if (casHead(h, s = snode(s, e, h, mode))) {
// 等待匹配，方法内部会尝试自旋、以及park操作
SNode m = awaitFulfill(s, timed, nanos);
// 处理匹配结果或取消情况
if (m == s) { // wait was cancelled
clean(s);
return null;
}
if ((h = head) != null && h.next == s)
casHead(h, s.next); // help s's fulfiller
return (E) ((mode == REQUEST) ? m.item : s.item);
}
// 尝试匹配不同模式的节点
} else if (!isFulfilling(h.mode)) { // try to fulfill
// 节点已取消
if (h.isCancelled()) // already cancelled
// 移除取消的节点并重试
casHead(h, h.next); // pop and retry
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
// 循环直到匹配成功或等待者消失
for (;;) { // loop until matched or waiters disappear
// 获取匹配节点
SNode m = s.next; // m is s's match
if (m == null) { // all waiters are gone
casHead(s, null); // pop fulfill node
s = null; // use new node next time
break; // restart main loop
}
SNode mn = m.next;
// 尝试匹配并处理结果
if (m.tryMatch(s)) {
casHead(s, mn); // pop both s and m
return (E) ((mode == REQUEST) ? m.item : s.item);
} else // lost match
s.casNext(m, mn); // help unlink
}
}
// 帮助一个正在匹配的线程
} else { // help a fulfiller
// 获取等待匹配的节点
SNode m = h.next; // m is h's match
// 协助完成匹配或移除已取消的节点
if (m == null) // waiter is gone
casHead(h, null); // pop fulfilling node
else {
SNode mn = m.next;
if (m.tryMatch(h)) // help match
casHead(h, mn); // pop both h and m
else // lost match
h.casNext(m, mn); // help unlink
}
}
}
}在整个过程中，TransferStack通过CAS操作来保证栈操作的原子性和线程安全，避免了使用传统锁机制带来的开销，提高了并发性能 。同时，通过巧妙的节点设计和模式判断，实现了生产者和消费者线程之间高效的数据传递和同步 。
TransferQueue（公平模式）
TransferQueue是SynchronousQueue在公平模式下的数据结构，它基于先进先出（FIFO）的队列结构来实现线程间的数据传递和同步，通过独特的节点设计和并发控制机制，确保了线程按照等待的先后顺序进行操作，体现了公平性原则 。
TransferQueue中的节点由QNode类表示，每个QNode节点包含了用于队列操作和线程同步的关键信息，其主要属性如下：
next：一个volatile修饰的QNode类型变量，指向队列中的下一个节点，用于构建队列的链表结构 。与TransferStack中的next指针类似，通过CAS操作来保证在多线程环境下对next指针的修改是线程安全的，确保队列的一致性 。item：volatile修饰的Object类型变量，用于存储节点的数据 。对于生产者节点，item存储要传递的数据；对于消费者节点，item为null，表示请求获取数据 。waiter：volatile修饰的Thread类型变量，保存了当前节点对应的等待线程 。当节点在队列中等待匹配时，对应的线程会被阻塞，通过waiter可以在匹配成功时唤醒该线程，实现线程间的协作 。isData：boolean类型变量，用于标识节点的类型 。如果isData为true，表示该节点是一个生产者数据节点；如果isData为false，表示该节点是一个消费者请求节点 。
TransferQueue的核心操作同样是transfer方法，它实现了公平模式下的数据转移和线程同步，其主要流程如下：
初始化与模式判断：首先，根据传入的参数
e判断当前操作的模式 。如果e为null，则表示当前是一个消费者线程，isData为false；如果e不为null，则表示当前是一个生产者线程，isData为true。同时，初始化一个QNode节点s，用于表示当前操作的线程 。自旋操作：进入一个无限循环（自旋），在每次循环中，获取当前队列的头节点
h和尾节点t。队列初始化检查：如果头节点
h或尾节点t为null，说明队列还未初始化完成，继续自旋等待，直到队列初始化完成 。这是为了确保在多线程环境下，队列的初始化操作能够正确完成，避免出现空指针异常等问题 。队列空或模式相同处理：如果头节点和尾节点相同（即队列空），或者尾节点的类型与当前操作模式相同（即都是生产者节点或都是消费者节点），则将当前节点
s添加到队列尾部 。首先，通过casNext方法使用CAS操作将尾节点的next指针指向当前节点s，如果CAS操作成功，说明添加节点成功，然后通过advanceTail方法将尾节点更新为当前节点s。接着，阻塞当前线程，等待匹配 。当前线程会通过awaitFulfill方法被挂起，该方法内部使用LockSupport.park方法实现线程的阻塞，直到被其他线程唤醒 。模式互补处理：如果队列不为空且尾节点的类型与当前操作模式互补（即一个是生产者节点，另一个是消费者节点），则表示找到了匹配的节点 。获取头节点的下一个节点
m（即等待匹配的节点），如果m的item与当前操作模式匹配（即生产者节点的item不为null且消费者节点的item为null），或者m已经被取消（m.item == m），或者m的item更新失败（!m.casItem(x, e)），则说明匹配出现问题，需要重新调整队列 。通过advanceHead方法将头节点更新为m，然后继续自旋 。如果匹配成功，将头节点的next指针指向m的next，实现节点的出队操作，同时通过LockSupport.unpark(m.waiter)方法唤醒m对应的等待线程 。最后，返回匹配节点中的数据（对于生产者线程，返回null；对于消费者线程，返回匹配节点的item） 。取消与清理操作：在等待过程中，如果当前线程被取消（例如通过
Thread.interrupted方法检测到中断），则会执行清理操作 。将当前节点从队列中移除，并唤醒队列中其他等待的线程 。具体操作是通过clean方法将当前节点的前驱节点的next指针指向当前节点的后继节点，从而绕过当前节点，实现节点的移除 。
在整个过程中，TransferQueue通过CAS操作和Condition条件变量来保证队列操作的原子性和线程安全，同时通过公平的队列结构和操作流程，确保了每个线程都能按照等待的先后顺序进行数据传递和同步，避免了线程饥饿问题，提供了稳定的并发性能 。
SynchronousQueue 关键方法源码解析
put 方法
put方法是SynchronousQueue中用于插入元素的核心方法之一，它的主要作用是将指定元素插入队列，但在SynchronousQueue中，由于其独特的无容量特性，插入操作必须等待有其他线程执行对应的移除操作才能完成。下面是put方法的源码实现：
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
if (transferer.transfer(e, false, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}从源码中可以看出，put方法首先对传入的元素e进行null值检查，如果e为null，则抛出NullPointerException，这是因为SynchronousQueue不允许插入null元素 。接着，put方法调用了transferer的transfer方法，transferer是SynchronousQueue内部的核心数据转移器，根据SynchronousQueue创建时指定的公平性模式，它可以是TransferStack（非公平模式）或TransferQueue（公平模式）的实例 。transfer方法的三个参数分别为：要插入的元素e、表示是否设置超时的布尔值timed（这里为false，表示无超时等待）、超时时间nanos（这里为0，因为无超时） 。
take 方法
take方法是SynchronousQueue中用于获取元素的关键方法，它的作用是从队列中取出一个元素，如果队列中没有可用元素，调用线程会被阻塞，直到有其他线程插入元素并与之匹配 。以下是take方法的源码：
public E take() throws InterruptedException {
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}take方法同样调用了transferer的transfer方法，与put方法不同的是，这里传入的第一个参数为null，表示当前是消费者操作，请求获取数据 。后面两个参数timed和nanos的含义与put方法中相同，这里表示无超时等待 。
offer 方法
offer方法用于尝试将元素插入SynchronousQueue，与put方法不同的是，offer方法不会一直阻塞等待，它有一个可设置的超时时间 。如果在指定的超时时间内无法将元素插入队列（即没有其他线程执行对应的移除操作），offer方法会返回false；如果插入成功，则返回true 。下面是offer方法的源码实现：
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
return true;
if (!Thread.interrupted())
return false;
throw new InterruptedException();
}offer方法首先对传入的元素e进行null值检查，若e为null，则抛出NullPointerException 。接着，调用transferer的transfer方法，与put方法相比，这里的第二个参数timed为true，表示设置了超时等待，第三个参数nanos则是将传入的超时时间timeout转换为纳秒后的数值 。
poll 方法
poll方法用于尝试从SynchronousQueue中取出一个元素，与take方法不同的是，poll方法也有一个可设置的超时时间 。如果在指定的超时时间内无法从队列中取出元素（即没有其他线程插入元素并与之匹配），poll方法会返回null；如果取出成功，则返回取出的元素 。以下是poll方法的源码：
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E e = transferer.transfer(null, true, unit.toNanos(timeout));
if (e != null || !Thread.interrupted())
return e;
throw new InterruptedException();
}poll方法调用transferer的transfer方法，传入的第一个参数为null，表示消费者操作，请求获取数据；第二个参数timed为true，表示设置了超时等待；第三个参数nanos是将传入的超时时间timeout转换为纳秒后的数值 。
SynchronousQueue 应用场景与案例分析
线程池中的应用
在 Java 线程池体系中，SynchronousQueue扮演着至关重要的角色，尤其在CachedThreadPool线程池中，其特性得到了淋漓尽致的发挥 。CachedThreadPool是一种可缓存的线程池，它的核心线程数为 0，最大线程数为Integer.MAX_VALUE，线程空闲时间为 60 秒，并且使用SynchronousQueue作为任务队列 。
当一个任务提交到CachedThreadPool时，SynchronousQueue的特性决定了任务的处理方式 。由于SynchronousQueue没有容量，它不会缓存任务，而是尝试直接将任务传递给一个可用的线程 。如果此时线程池中有空闲线程，任务会立即被该线程执行；如果没有空闲线程，线程池会创建一个新的线程来执行任务 。这种机制确保了任务能够得到即时处理，避免了任务在队列中的积压，特别适合处理大量短生命周期的任务 。
以一个简单的 Web 服务器场景为例，假设我们有一个处理 HTTP 请求的应用，使用CachedThreadPool和SynchronousQueue来处理请求 。当客户端发起 HTTP 请求时，请求会被封装成任务提交到线程池 。由于SynchronousQueue的特性，请求任务会迅速被分配到空闲线程或者新创建的线程中进行处理 。如果在高并发情况下，大量请求涌入，SynchronousQueue不会将请求堆积起来，而是促使线程池及时创建新线程来应对，保证了系统对请求的快速响应 。
在代码实现上，CachedThreadPool的创建和任务提交非常简洁：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CachedThreadPoolDemo {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int taskNumber = i;
executorService.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is handling task " + taskNumber);
try {
Thread.sleep(1000); // 模拟任务处理时间
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
}在这段代码中，Executors.newCachedThreadPool()创建了一个CachedThreadPool实例，其中的SynchronousQueue负责任务的传递 。当循环提交 10 个任务时，每个任务都会尽快被线程池中的线程处理，充分体现了SynchronousQueue在这种场景下的高效性 。
然而，SynchronousQueue在CachedThreadPool中的应用也存在一些潜在问题 。由于线程池的最大线程数理论上是无限的，如果任务提交速度过快，可能会导致系统创建大量线程，从而耗尽系统资源，如内存和 CPU 。因此，在使用CachedThreadPool和SynchronousQueue时，需要根据实际业务场景和系统资源状况，合理控制任务的提交速率和线程池的使用 。
生产者 - 消费者场景
在生产者 - 消费者模型中，SynchronousQueue以其独特的同步机制，为数据的传递提供了一种高效且精准的方式 。下面通过一个具体的案例来深入理解它在该场景中的应用 。
假设我们正在开发一个视频处理系统，其中生产者线程负责从视频源读取视频帧数据，消费者线程负责对这些视频帧进行图像处理，如视频剪辑、特效添加等 。在这个系统中，使用SynchronousQueue来传递视频帧数据 。
import java.util.concurrent.SynchronousQueue;
public class ProducerConsumerExample {
public static void main(String[] args) {
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
Thread producer = new Thread(() -> {
try {
for (int i = 1; i <= 5; i++) {
System.out.println("Producer produced frame: " + i);
queue.put(i); // 生产者将视频帧放入队列
Thread.sleep(1000); // 模拟生产间隔
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
Thread consumer = new Thread(() -> {
try {
for (int i = 1; i <= 5; i++) {
Integer frame = queue.take(); // 消费者从队列中取出视频帧
System.out.println("Consumer consumed frame: " + frame);
Thread.sleep(1500); // 模拟消费间隔
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
producer.start();
consumer.start();
}
}在这个示例中，生产者线程按顺序生成 5 个视频帧，并通过queue.put(i)将其放入SynchronousQueue中 。由于SynchronousQueue没有容量，生产者线程在放入每个视频帧后会被阻塞，直到消费者线程通过queue.take()取走该视频帧 。消费者线程取出视频帧后，进行相应的处理，并模拟了较长的处理时间 。这种严格的同步机制确保了视频帧的即时传递和处理，避免了数据的积压和丢失，保证了视频处理的实时性和连贯性 。
从适用场景来看，SynchronousQueue非常适合生产者和消费者处理速度相近，且对数据传递的实时性要求极高的场景 。在这种场景下，SynchronousQueue的直接移交特性能够避免数据在队列中的等待时间，提高系统的整体性能 。然而，它也存在一些局限性 。如果生产者和消费者的处理速度差异较大，可能会导致其中一方长时间等待，影响系统的效率 。比如在上述视频处理案例中，如果消费者处理视频帧的速度远低于生产者生成视频帧的速度，生产者线程就会频繁地被阻塞等待，降低了系统的吞吐量 。
为了应对这种情况，在实际应用中需要根据生产者和消费者的性能特点，合理调整它们的处理逻辑和资源分配 。例如，可以通过优化消费者的处理算法，提高其处理速度；或者在生产者端添加一定的缓冲机制，当消费者处理速度较慢时，暂时存储一些数据，避免生产者长时间等待 。同时，还可以结合其他并发工具和设计模式，如线程池、信号量等，来进一步优化生产者 - 消费者模型的性能和稳定性 。
SynchronousQueue 使用注意事项与最佳实践
避免死锁
在使用SynchronousQueue时，死锁是一个需要特别关注的问题，尤其是在单线程环境或使用不当的情况下，死锁的风险会显著增加。
在单线程环境中，如果一个线程在SynchronousQueue上执行put操作，由于没有其他线程来执行对应的take操作，该线程将永远阻塞，从而导致死锁 。例如：
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
new Thread(() -> {
try {
queue.put(1); // 没有消费者线程，此操作将永久阻塞
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();在上述代码中，put操作会一直等待take操作，而由于没有消费者线程，这种等待会一直持续下去，最终导致死锁 。
在多线程环境中，如果线程之间的操作顺序不当，也可能引发死锁 。比如，在生产者 - 消费者模型中，如果生产者线程在put操作前持有某个锁，而消费者线程在take操作前也需要获取这个锁，并且获取顺序相反，就可能出现死锁 。假设我们有一个资源resource，生产者和消费者都需要先获取resource的锁，然后再对SynchronousQueue进行操作：
class Resource {
// 模拟资源
}
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
Resource resource = new Resource();
Thread producer = new Thread(() -> {
synchronized (resource) {
try {
queue.put(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread consumer = new Thread(() -> {
synchronized (resource) {
try {
queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
consumer.start();在这种情况下，如果生产者线程先获取了resource的锁，然后执行put操作被阻塞，等待消费者线程执行take操作；而消费者线程此时也在等待获取resource的锁，因为生产者线程持有锁而无法获取，从而形成死锁 。
为了避免死锁，首先要确保在使用SynchronousQueue时，有足够的线程来执行put和take操作，以保证它们能够相互匹配 。在单线程场景下，应避免使用SynchronousQueue进行同步操作，或者确保有其他机制来触发相应的操作 。在多线程环境中，要谨慎设计线程之间的同步逻辑和锁的获取顺序，尽量避免嵌套锁的使用，或者采用死锁检测和恢复机制，如使用ThreadMXBean来检测死锁，并在检测到死锁时采取相应的恢复措施，如终止相关线程或释放锁资源 。同时，可以使用offer和poll方法，并设置合理的超时时间，避免线程无限期阻塞 。例如：
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
boolean success = queue.offer(1, 5, TimeUnit.SECONDS); // 尝试在5秒内插入元素
if (!success) {
// 处理插入失败的情况
}通过设置超时时间，如果在规定时间内无法完成操作，程序可以采取其他策略，从而避免死锁的发生 。
内存管理
在使用SynchronousQueue时，内存管理是一个容易被忽视但又至关重要的问题。由于SynchronousQueue本身不存储元素，它的内存占用主要来自于等待线程所占用的资源以及内部数据结构的开销 。然而，在一些特殊情况下，如线程被中断时未正确处理，可能会导致内存泄漏等问题 。
当线程在SynchronousQueue上执行take或put操作时，如果被中断，需要正确处理中断异常，以避免资源的无效占用 。例如，在下面的代码中：
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
Thread thread = new Thread(() -> {
try {
queue.take();
} catch (InterruptedException e) {
// 未正确处理中断，可能导致内存泄漏
}
});
thread.start();
// 模拟中断线程
thread.interrupt();在这个例子中，当线程被中断时，catch块中没有对中断进行任何处理，这可能导致线程虽然被中断，但仍然占用着SynchronousQueue内部的数据结构资源，如TransferStack或TransferQueue中的节点，从而造成内存泄漏 。
为了避免这种情况，在捕获到InterruptedException时，应该根据具体业务逻辑进行适当的处理 。通常，可以选择恢复中断状态，以便上层调用者能够知晓线程被中断的情况 。例如：
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
Thread thread = new Thread(() -> {
try {
queue.take();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // 恢复中断状态
// 可以根据业务需求进行其他处理，如记录日志等
}
});
thread.start();
// 模拟中断线程
thread.interrupt();此外，在使用SynchronousQueue时，要注意线程池的配置 。如果使用SynchronousQueue作为线程池的任务队列，线程池的线程数量设置不当可能会导致大量线程被阻塞，从而占用过多内存 。在CachedThreadPool中，由于其使用SynchronousQueue且最大线程数理论上是无限的，如果任务提交速度过快，可能会创建大量线程，耗尽内存资源 。因此，需要根据实际业务负载，合理设置线程池的核心线程数和最大线程数，确保系统能够高效稳定运行，同时避免内存资源的浪费和过度占用 。
性能优化
在使用SynchronousQueue时，合理的性能优化策略能够充分发挥其高效的线程间协作能力，提升系统的整体性能 。
公平模式的设置对SynchronousQueue的性能有着显著影响 。公平模式下，SynchronousQueue使用TransferQueue，按照先进先出（FIFO）的顺序处理线程请求，这确保了每个线程都有公平的机会进行操作，避免了线程饥饿问题 。在一些对公平性要求较高的场景，如实时控制系统中，使用公平模式可以保证任务的稳定执行 。然而，公平模式由于需要维护严格的顺序，在高并发情况下，线程切换和队列操作的开销相对较大，可能会影响系统的吞吐量 。非公平模式下，SynchronousQueue使用TransferStack，采用后进先出（LIFO）的栈结构，在高并发场景下，由于减少了线程切换的开销，能够在一定程度上提高系统的吞吐量，但可能会导致某些线程长时间等待，出现 “饥饿” 现象 。因此，在选择公平模式还是非公平模式时，需要根据具体的业务场景和性能需求进行权衡 。如果业务场景对公平性要求较高，且并发量不是特别大，公平模式是一个不错的选择；如果业务场景追求高吞吐量，且对公平性要求相对较低，非公平模式可能更适合 。
配合线程池参数进行优化也是提升性能的关键 。当SynchronousQueue作为线程池的任务队列时，线程池的核心线程数和最大线程数的设置至关重要 。在CachedThreadPool中，由于使用SynchronousQueue且最大线程数为Integer.MAX_VALUE，如果任务提交速度过快，可能会创建大量线程，耗尽系统资源 。因此，在使用SynchronousQueue的线程池时，需要根据实际业务负载，合理设置线程池的参数 。如果任务处理时间较短且并发量较大，可以适当增加核心线程数，减少线程创建和销毁的开销；如果任务处理时间较长，需要控制最大线程数，避免资源耗尽 。同时，还可以结合线程池的拒绝策略，如CallerRunsPolicy，当线程池和队列都满载时，让提交任务的线程直接执行任务，避免任务被拒绝，从而保证系统的稳定性和性能 。例如，在一个处理 HTTP 请求的应用中，根据服务器的硬件配置和预估的并发请求量，合理设置线程池的核心线程数为 10，最大线程数为 50，并使用SynchronousQueue作为任务队列，同时采用CallerRunsPolicy拒绝策略，能够有效地处理大量并发请求，提高系统的响应速度和吞吐量 。
总结
SynchronousQueue作为 Java 并发包中一种独特的阻塞队列，具有无容量、线程间直接移交元素的显著特点，其内部基于Transferer接口实现了两种不同的数据结构：非公平模式下的TransferStack和公平模式下的TransferQueue 。这种设计使得SynchronousQueue在不同的应用场景中都能发挥出高效的线程间协作能力 。
在性能方面，SynchronousQueue的无锁化设计和基于 CAS 的操作机制，使其在高并发场景下能够实现极低的延迟和极高的吞吐量 。在公平模式下，TransferQueue通过 FIFO 的队列结构保证了线程的公平性，适用于对公平性要求较高的场景，如实时控制系统；在非公平模式下，TransferStack采用 LIFO 的栈结构，减少了线程切换的开销，能够在一定程度上提高系统的吞吐量，更适合高并发服务器等对吞吐量要求较高的场景 。
在应用场景上，SynchronousQueue在 Java 线程池和生产者 - 消费者模型中有着广泛的应用 。在CachedThreadPool线程池中，SynchronousQueue确保了任务的即时处理，避免了任务在队列中的积压，特别适合处理大量短生命周期的任务 。在生产者 - 消费者场景中，它实现了数据的即时传递，保证了数据处理的实时性和连贯性 。
