[toc]
在前面学习过java线程间通信的几种方式，分别是synchronized结合wait/notify。以及ReentrantLock结合Condation的await和signal方法。那么实际上还有一种方法也能实现线程通信，那就是LockSupport。我们来看看其源码。
1.类注释说明
在类源码中有大段注释，在前面学习的过程中不难发现，凡是java代码中的注释，都是非常重要的。
/**
* Basic thread blocking primitives for creating locks and other
* synchronization classes.
*
* <p>This class associates, with each thread that uses it, a permit
* (in the sense of the {@link java.util.concurrent.Semaphore
* Semaphore} class). A call to {@code park} will return immediately
* if the permit is available, consuming it in the process; otherwise
* it <em>may</em> block. A call to {@code unpark} makes the permit
* available, if it was not already available. (Unlike with Semaphores
* though, permits do not accumulate. There is at most one.)
*
* <p>Methods {@code park} and {@code unpark} provide efficient
* means of blocking and unblocking threads that do not encounter the
* problems that cause the deprecated methods {@code Thread.suspend}
* and {@code Thread.resume} to be unusable for such purposes: Races
* between one thread invoking {@code park} and another thread trying
* to {@code unpark} it will preserve liveness, due to the
* permit. Additionally, {@code park} will return if the caller's
* thread was interrupted, and timeout versions are supported. The
* {@code park} method may also return at any other time, for "no
* reason", so in general must be invoked within a loop that rechecks
* conditions upon return. In this sense {@code park} serves as an
* optimization of a "busy wait" that does not waste as much time
* spinning, but must be paired with an {@code unpark} to be
* effective.
*
* <p>The three forms of {@code park} each also support a
* {@code blocker} object parameter. This object is recorded while
* the thread is blocked to permit monitoring and diagnostic tools to
* identify the reasons that threads are blocked. (Such tools may
* access blockers using method {@link #getBlocker(Thread)}.)
* The use of these forms rather than the original forms without this
* parameter is strongly encouraged. The normal argument to supply as
* a {@code blocker} within a lock implementation is {@code this}.
*
* <p>These methods are designed to be used as tools for creating
* higher-level synchronization utilities, and are not in themselves
* useful for most concurrency control applications. The {@code park}
* method is designed for use only in constructions of the form:
*
* <pre> {@code
* while (!canProceed()) { ... LockSupport.park(this); }}</pre>
*
* where neither {@code canProceed} nor any other actions prior to the
* call to {@code park} entail locking or blocking. Because only one
* permit is associated with each thread, any intermediary uses of
* {@code park} could interfere with its intended effects.
*
* <p><b>Sample Usage.</b> Here is a sketch of a first-in-first-out
* non-reentrant lock class:
* <pre> {@code
* class FIFOMutex {
* private final AtomicBoolean locked = new AtomicBoolean(false);
* private final Queue<Thread> waiters
* = new ConcurrentLinkedQueue<Thread>();
*
* public void lock() {
* boolean wasInterrupted = false;
* Thread current = Thread.currentThread();
* waiters.add(current);
*
* // Block while not first in queue or cannot acquire lock
* while (waiters.peek() != current ||
* !locked.compareAndSet(false, true)) {
* LockSupport.park(this);
* if (Thread.interrupted()) // ignore interrupts while waiting
* wasInterrupted = true;
* }
*
* waiters.remove();
* if (wasInterrupted) // reassert interrupt status on exit
* current.interrupt();
* }
*
* public void unlock() {
* locked.set(false);
* LockSupport.unpark(waiters.peek());
* }
* }}</pre>
*/
上面的注释的大意为:
LockSupport是一个创建锁和其他同步类的线程阻塞原语。
在个类在使用其的每个线程上关联一个许可类（ java.util.concurrent.Semaphore）。如果许可证可用，对park调用将立即返回，并在进程中使用它，否则就会阻塞。当许可证不可用的情况下调用unpark方法则可以使许可证可用。与信号量Semaphore不同的是，许可证不会累计，最多只有一个。
方法park和unpark提供了有效的阻塞和解除阻塞线程的功能。但是不能与已废弃的方法suspend和resume方法一样，用于如下目的：在一个调用park的线程和另外一个调用unpark的线程之间对许可证进行竞争，保持活力。另外，park在调用方线程被中断的时候返回。并且支持超时。park方法也可能在任何其他的时候返回，因为没有原因，因此通常必须在返回时重新检查条件中循环调用。从这个意义上说，park是繁忙等待的优化。它不会浪费太多的时间自旋，但是必须与unpark配合使用才有效。
park有三种调用形式，每种都支持blocker对象参数，此对象在线程被阻塞的时候被记录，以便监视和诊断工具能够识别线程被阻塞的原因。这类工具可以用getBlocker的方式访问被阻塞队程序，强烈建议使用带有bloker参数的方法，而不是不带这些参数的原始方法。在锁实现中做为blocker提供的常规参数是this。
这些方法被设计用来做为创建更高级别的同步工具。而不是自己在对大多数并发控制程英语程序中使用，park方法仅仅用于下列形式的构造：
while (!canProceed()) {
...
LockSupport.park(this);
}}
在canProcess方法或者任何其他操作之前调用park需要加锁或者阻塞，因为只有一个许可证，这是与每个线程都关联的，任何中间方使用park都可能对park的预期效果造成影响。
如下，这是一个先进先出的不可重入的锁的伪代码：
class FIFOMutex {
private final AtomicBoolean locked = new AtomicBoolean(false);
private final Queue<Thread> waiters
= new ConcurrentLinkedQueue<Thread>();
public void lock() {
boolean wasInterrupted = false;
Thread current = Thread.currentThread();
waiters.add(current);
// Block while not first in queue or cannot acquire lock
while (waiters.peek() != current ||
!locked.compareAndSet(false, true)) {
LockSupport.park(this);
if (Thread.interrupted()) // ignore interrupts while waiting
wasInterrupted = true;
}
waiters.remove();
if (wasInterrupted) // reassert interrupt status on exit
current.interrupt();
}
public void unlock() {
locked.set(false);
LockSupport.unpark(waiters.peek());
}
2.主要方法
2.1 park
park我们可以形象的理解为泊车的意思，实际上，park就是一个标识0-1的信号的共享变量。如果park，则标识占用。那么信号就变成0.如果执行unpark则信号变成1。
/**
* Disables the current thread for thread scheduling purposes unless the
* permit is available.
*
* <p>If the permit is available then it is consumed and the call
* returns immediately; otherwise the current thread becomes disabled
* for thread scheduling purposes and lies dormant until one of three
* things happens:
*
* <ul>
*
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread upon return.
*/
public static void park() {
UNSAFE.park(false, 0L);
}
park实际上使用的是UNSAFE类里面的park和unpark操作，而LocalSupport仅仅是对UnSafe这个类的功能进行的一些列的封装。
其注释大意为：
在线程调度的时候禁用当前线程，将其放置到WaitSet队列。除非有许可证。所谓的许可证就是前面说的0-1的标识。使用unpark则就会产生一个许可。
如果许可可用，那么使用它并立即返回，否则大昂区线程将被线程调度禁用，处于休眠状态，除非有如下之一的情况发生：
- 其他线程以当前线程为目标调用unpark方法。
- 其他线程打断当前线程。
- 调用了虚假的返回。
该方法不会报告是按个原因导致的方法返回，调用者再次检查造成的原因需要park这个线程，调用者还可以决定线程返回时的中断状态。
2.2 park(Object blocker)
/**
* Disables the current thread for thread scheduling purposes unless the
* permit is available.
*
* <p>If the permit is available then it is consumed and the call returns
* immediately; otherwise
* the current thread becomes disabled for thread scheduling
* purposes and lies dormant until one of three things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @since 1.6
*/
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
这个方法与park方法类似。但是可以传入一个blocker参数，这个参数表示负责此操作的同步对象的线程。
这个方法与park方法的本质是一样的。只是在调用UNSAFE的park方法之前通过setBlocker方法传入的blocker的表示在当前的线程的内存对象中。
private static void setBlocker(Thread t, Object arg) {
// Even though volatile, hotspot doesn't need a write barrier here.
UNSAFE.putObject(t, parkBlockerOffset, arg);
}
同样，setBlocker也是采用UNSAFE.putObject来实现的。
2.3 parkNanos(Object blocker, long nanos)
/**
* Disables the current thread for thread scheduling purposes, for up to
* the specified waiting time, unless the permit is available.
*
* <p>If the permit is available then it is consumed and the call
* returns immediately; otherwise the current thread becomes disabled
* for thread scheduling purposes and lies dormant until one of four
* things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The specified waiting time elapses; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread, or the elapsed time
* upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @param nanos the maximum number of nanoseconds to wait
* @since 1.6
*/
public static void parkNanos(Object blocker, long nanos) {
if (nanos > 0) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, nanos);
setBlocker(t, null);
}
}
此方法与park方法类似，但是会传入一个过期时间，这个过期时间是以纳秒为单位的。
需要注意的是这个方法如果传入的nanos小于等于0则该方法不会生效。
UNSAFE.park(false, nanos);
可见park方法的第二个参数单位是纳秒。
2.4 parkUntil(Object blocker, long deadline)
/**
* Disables the current thread for thread scheduling purposes, until
* the specified deadline, unless the permit is available.
*
* <p>If the permit is available then it is consumed and the call
* returns immediately; otherwise the current thread becomes disabled
* for thread scheduling purposes and lies dormant until one of four
* things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts} the
* current thread; or
*
* <li>The specified deadline passes; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread, or the current time
* upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @param deadline the absolute time, in milliseconds from the Epoch,
* to wait until
* @since 1.6
*/
public static void parkUntil(Object blocker, long deadline) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(true, deadline);
setBlocker(t, null);
}
此方法同样与park方法没有本质的区别，只是多加了一个过期时间的条件。这个过期时间以毫秒为单位。实际上UNSAFE的park方法在这个方法与上个方法之间的差别还有要给就是第一个参数在此处为true。
那么可以猜想就是为true的时候以毫秒为超时单位。为false的时候超时的单位就是纳秒。
2.5 unpark
unpark方法则是之前park方法的授予许可证的过程，如果有许可，那么park调用的时候不会阻塞。
/**
* Makes available the permit for the given thread, if it
* was not already available. If the thread was blocked on
* {@code park} then it will unblock. Otherwise, its next call
* to {@code park} is guaranteed not to block. This operation
* is not guaranteed to have any effect at all if the given
* thread has not been started.
*
* @param thread the thread to unpark, or {@code null}, in which case
* this operation has no effect
*/
public static void unpark(Thread thread) {
if (thread != null)
UNSAFE.unpark(thread);
}
调用unpark之后，会将标识变为0，之后park就会被阻塞。
需要注意的是，实际上LocakSupport比较灵活。与wait/notify机制不同的是，其不需要约定先后顺序。也就是说，可以先调用unpark方法。这对于之前wait/notify绝对是巨大的改善。因为wait方法的时候，如果先使用notify，将没有任何效果。一旦用错顺序可以会导致死锁。
3.应用测试
3.1 测试 interrupt
public class LockSupportTest2 {
public static void main(String[] args) throws InterruptedException{
Thread t1 = new Thread(new T1(),"T1");
t1.start();
TimeUnit.SECONDS.sleep(1);
t1.interrupt();
LockSupport.unpark(t1);
}
static class T1 implements Runnable {
@Override
public void run() {
while (true) {
System.out.println(Thread.currentThread().getName()+": print");
LockSupport.park();
}
}
}
}
上述代码执行结果：
T1: begin loop
T1: print
T1: begin loop
T1: print
...
上述代码会循环执行。不会停止。这是因为，当线程调用interrupt之后，线程就处于interrupt状态。之后的park方法则不会生效。实际上park在interrupt之后不会出现任何异常。
3.2 park & unpark
有了park和unpark之后，我们再回到之前的例子，需要两个线程交替打印1-20，那么这就将非常简单：
public class LockSuppotTest {
private static volatile int count = 0;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while (true) {
System.out.println(Thread.currentThread().getName() + ":" + count++);
LockSupport.park();
}
},"T1");
Thread t2 = new Thread(() -> {
while (true) {
System.out.println(Thread.currentThread().getName() + ":" + count++);
LockSupport.park();
}
},"T2");
Thread t3 = new Thread(() -> {
try {
for(int i=0;i<19;i++){
TimeUnit.SECONDS.sleep(1);
if((i&1) == 0) {
LockSupport.unpark(t1);
}else {
LockSupport.unpark(t2);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T3");
t1.setDaemon(true);
t2.setDaemon(true);
t1.start();
TimeUnit.MILLISECONDS.sleep(100);
t2.start();
t3.start();
t3.join();
System.out.println("exit!");
}
}
上述代码执行结果：
T1:0
T2:1
T1:2
T2:3
T1:4
T2:5
T1:6
T2:7
T1:8
T2:9
T1:10
T2:11
T1:12
T2:13
T1:14
T2:15
T1:16
T2:17
T1:18
T2:19
T1:20
exit!
这样就实现了一个不需要通过synchronized或者ReentrantLock的交替打印的例子。
4.总结
LocalSupport也是实现线程间通信的一种有效的方式。而且非常灵活。unpark和park之间不需要有严格的顺序。可以先执行unpark，之后再执行park。这样再编码的过程中就比使用wait/notify方法要简单很多。
此外，LocalSUpport全部是采用UnSafe类来实现的。这个类通过使用park/unpark以及相关cas操作，就实现了java中JUC的各种复杂的数据结构和容器。而且效率非常高。
因此UnSafe及CAS也是后面需要重点说明的部分。
网友评论