concurrent包是基于AQS (AbstractQueuedSynchronizer)框架的&＃xff0c;AQS(JAVA CAS原理、unsafe、AQS)框架借助于两个类&＃xff1a;
- Unsafe&＃xff08;提供CAS操作&＃xff09;------------------------------------《AQS之&＃xff1a;Unsafe&＃xff08;java可直接操作内存()&＃xff0c;挂起与恢复&＃xff0c;CAS操作&＃xff09;》
- LockSupport&＃xff08;提供park/unpark操作&＃xff09;-------------------《Java多线程系列--AQS之 LockSupport》
因此&＃xff0c;LockSupport非常重要。
LockSupport介绍
LockSupport是JDK中比较底层的类&＃xff0c;用来创建锁和其他同步工具类的基本线程阻塞原语。java锁和同步器框架的核心 AQS: AbstractQueuedSynchronizer&＃xff0c;就是通过调用 LockSupport .park()和 LockSupport .unpark()实现线程的阻塞和解除阻塞的。LockSupport中的park() 和 unpark() 的作用分别是阻塞线程和解除阻塞线程&＃xff0c;而且park()和unpark()不会遇到“Thread.suspend 和 Thread.resume所可能引发的死锁”问题。
LockSupport类是Java6(JSR166-JUC)引入的一个类&＃xff0c;提供了基本的线程同步原语。LockSupport实际上是调用了Unsafe类里的函数&＃xff0c;归结到Unsafe里&＃xff0c;只有两个函数&＃xff1a;
- park&＃xff1a;阻塞当前线程(Block current thread),字面理解park&＃xff0c;就算占住&＃xff0c;停车的时候不就把这个车位给占住了么&＃xff1f;起这个名字还是很形象的。
- unpark: 使给定的线程停止阻塞(Unblock the given thread blocked )。
因为park() 和 unpark()有许可的存在&＃xff1b;调用 park() 的线程和另一个试图将其 unpark() 的线程之间的竞争将保持活性。
类图&＃xff1a;
LockSupport函数列表
//
private static void setBlocker(Thread t, Object arg)
// 返回提供给最近一次尚未解除阻塞的 park 方法调用的 blocker 对象&＃xff0c;如果该调用不受阻塞&＃xff0c;则返回 null。
static Object getBlocker(Thread t)
// 为了线程调度&＃xff0c;禁用当前线程&＃xff0c;除非许可可用。
static void park()
// 为了线程调度&＃xff0c;在许可可用之前禁用当前线程。
static void park(Object blocker)
// 为了线程调度禁用当前线程&＃xff0c;最多等待指定的等待时间&＃xff0c;除非许可可用。
static void parkNanos(long nanos)
// 为了线程调度&＃xff0c;在许可可用前禁用当前线程&＃xff0c;并最多等待指定的等待时间。
static void parkNanos(Object blocker, long nanos)
// 为了线程调度&＃xff0c;在指定的时限前禁用当前线程&＃xff0c;除非许可可用。
static void parkUntil(long deadline)
// 为了线程调度&＃xff0c;在指定的时限前禁用当前线程&＃xff0c;除非许可可用。
static void parkUntil(Object blocker, long deadline)
// 如果给定线程的许可尚不可用&＃xff0c;则使其可用。
static void unpark(Thread thread)
两个重点
(1)操作对象
归根结底&＃xff0c;LockSupport调用的Unsafe中的native代码&＃xff1a;
public static void unpark(Thread thread) {if (thread !&＃61; null)UNSAFE.unpark(thread);}public static void park() {UNSAFE.park(false, 0L);}
两个函数声明清楚地说明了操作对象&＃xff1a;park函数是将当前Thread阻塞&＃xff0c;而unpark函数则是将指定线程Thread唤醒。
与Object类的wait/notify机制相比&＃xff0c;park/unpark有两个优点&＃xff1a;
1. 以thread为操作对象更符合阻塞线程的直观定义&＃xff1b;
2. 操作更精准&＃xff0c;可以准确地唤醒某一个线程&＃xff08;notify随机唤醒一个线程&＃xff0c;notifyAll唤醒所有等待的线程&＃xff09;&＃xff0c;增加了灵活性。
(2)关于许可
在上面的文字中&＃xff0c;我使用了阻塞和唤醒&＃xff0c;是为了和wait/notify做对比。
- 其实park/unpark的设计原理核心是“许可”。park是等待一个许可。unpark是为某线程提供一个许可。如果某线程A调用park&＃xff0c;那么除非另外一个线程调用unpark(A)给A一个许可&＃xff0c;否则线程A将阻塞在park操作上。
- 有一点比较难理解的&＃xff0c;是unpark操作可以再park操作之前。也就是说&＃xff0c;先提供许可。当某线程调用park时&＃xff0c;已经有许可了&＃xff0c;它就消费这个许可&＃xff0c;然后可以继续运行。这其实是必须的。考虑最简单的生产者(Producer)消费者(Consumer)模型&＃xff1a;Consumer需要消费一个资源&＃xff0c;于是调用park操作等待&＃xff1b;Producer则生产资源&＃xff0c;然后调用unpark给予Consumer使用的许可。非常有可能的一种情况是&＃xff0c;Producer先生产&＃xff0c;这时候Consumer可能还没有构造好&＃xff08;比如线程还没启动&＃xff0c;或者还没切换到该线程&＃xff09;。那么等Consumer准备好要消费时&＃xff0c;显然这时候资源已经生产好了&＃xff0c;可以直接用&＃xff0c;那么park操作当然可以直接运行下去。如果没有这个语义&＃xff0c;那将非常难以操作。
- 但是这个“许可”是不能叠加的&＃xff0c;“许可”是一次性的。比如线程B连续调用了三次unpark函数&＃xff0c;当线程A调用park函数就使用掉这个“许可”&＃xff0c;如果线程A再次调用park&＃xff0c;则进入等待状态。
park和unpark的灵活之处
上面已经提到&＃xff0c;unpark函数可以先于park调用&＃xff0c;这个正是它们的灵活之处。
一个线程它有可能在别的线程unPark之前&＃xff0c;或者之后&＃xff0c;或者同时调用了park&＃xff0c;那么因为park的特性&＃xff0c;它可以不用担心自己的park的时序问题&＃xff0c;否则&＃xff0c;如果park必须要在unpark之前&＃xff0c;那么给编程带来很大的麻烦&＃xff01;&＃xff01;
考虑一下&＃xff0c;两个线程同步&＃xff0c;要如何处理&＃xff1f;
在Java5里是用wait/notify/notifyAll来同步的。wait/notify机制有个很蛋疼的地方是&＃xff0c;比如线程B要用notify通知线程A&＃xff0c;那么线程B要确保线程A已经在wait调用上等待了&＃xff0c;否则线程A可能永远都在等待。编程的时候就会很蛋疼。
另外&＃xff0c;是调用notify&＃xff0c;还是notifyAll&＃xff1f;
notify只会唤醒一个线程&＃xff0c;如果错误地有两个线程在同一个对象上wait等待&＃xff0c;那么又悲剧了。为了安全起见&＃xff0c;貌似只能调用notifyAll了。而unpark可以指定到特定线程。
park/unpark模型真正解耦了线程之间的同步&＃xff0c;线程之间不再需要一个Object或者其它变量来存储状态&＃xff0c;不再需要关心对方的状态。
Unsafe.park和Unsafe.unpark的底层实现原理
在Linux系统下&＃xff0c;是用的Posix线程库pthread中的mutex&＃xff08;互斥量&＃xff09;&＃xff0c;condition&＃xff08;条件变量&＃xff09;来实现的。
mutex和condition保护了一个_counter的变量&＃xff0c;当park时&＃xff0c;这个变量被设置为0&＃xff0c;当unpark时&＃xff0c;这个变量被设置为1。
每个Java线程都有一个Parker实例&＃xff0c;Parker类是这样定义的&＃xff1a;
class Parker : public os::PlatformParker {
private: volatile int _counter ; ...
public: void park(bool isAbsolute, jlong time); void unpark(); ...
}
class PlatformParker : public CHeapObj
}
可以看到Parker类实际上用Posix的mutex&＃xff0c;condition来实现的。
在Parker类里的_counter字段&＃xff0c;就是用来记录“许可”的。
- park 过程
当调用park时&＃xff0c;先尝试能否直接拿到“许可”&＃xff0c;即_counter>0时&＃xff0c;如果成功&＃xff0c;则把_counter设置为0&＃xff0c;并返回&＃xff1a;
void Parker::park(bool isAbsolute, jlong time) { // Ideally we&＃39;d do something useful while spinning, such // as calling unpackTime(). // Optional fast-path check: // Return immediately if a permit is available. // We depend on Atomic::xchg() having full barrier semantics // since we are doing a lock-free update to _counter. if (Atomic::xchg(0, &_counter) > 0) return;
如果不成功&＃xff0c;则构造一个ThreadBlockInVM&＃xff0c;然后检查_counter是不是>0&＃xff0c;如果是&＃xff0c;则把_counter设置为0&＃xff0c;unlock mutex并返回&＃xff1a;
ThreadBlockInVM tbivm(jt);
if (_counter > 0) { // no wait needed _counter &＃61; 0; status &＃61; pthread_mutex_unlock(_mutex);
否则&＃xff0c;再判断等待的时间&＃xff0c;然后再调用pthread_cond_wait函数等待&＃xff0c;如果等待返回&＃xff0c;则把_counter设置为0&＃xff0c;unlock mutex并返回&＃xff1a;
if (time &＃61;&＃61; 0) { status &＃61; pthread_cond_wait (_cond, _mutex) ;
}
_counter &＃61; 0 ;
status &＃61; pthread_mutex_unlock(_mutex) ;
assert_status(status &＃61;&＃61; 0, status, "invariant") ;
OrderAccess::fence();
- unpark 过程
当unpark时&＃xff0c;则简单多了&＃xff0c;直接设置_counter为1&＃xff0c;再unlock mutex返回。如果_counter之前的值是0&＃xff0c;则还要调用pthread_cond_signal唤醒在park中等待的线程&＃xff1a;
void Parker::unpark() { int s, status ; status &＃61; pthread_mutex_lock(_mutex); assert (status &＃61;&＃61; 0, "invariant") ; s &＃61; _counter; _counter &＃61; 1; if (s <1) { if (WorkAroundNPTLTimedWaitHang) { status &＃61; pthread_cond_signal (_cond) ; assert (status &＃61;&＃61; 0, "invariant") ; status &＃61; pthread_mutex_unlock(_mutex); assert (status &＃61;&＃61; 0, "invariant") ; } else { status &＃61; pthread_mutex_unlock(_mutex); assert (status &＃61;&＃61; 0, "invariant") ; status &＃61; pthread_cond_signal (_cond) ; assert (status &＃61;&＃61; 0, "invariant") ; } } else { pthread_mutex_unlock(_mutex); assert (status &＃61;&＃61; 0, "invariant") ; }
}
LockSupport示例
对比下面的“示例1”和“示例2”可以更清晰的了解LockSupport的用法。
示例1
package lock.demo7;public class WaitTest1 {public static void main(String[] args) {ThreadA ta &＃61; new ThreadA("ta");synchronized (ta) { // 通过synchronized(ta)获取“对象ta的同步锁”try {System.out.println(Thread.currentThread().getName() &＃43; " start ta");ta.start();System.out.println(Thread.currentThread().getName() &＃43; " block");// 主线程等待
ta.wait();System.out.println(Thread.currentThread().getName() &＃43; " continue");} catch (InterruptedException e) {e.printStackTrace();}}}static class ThreadA extends Thread {public ThreadA(String name) {super(name);}public void run() {synchronized (this) { // 通过synchronized(this)获取“当前对象的同步锁”System.out.println(Thread.currentThread().getName() &＃43; " wakup others");notify(); // 唤醒“当前对象上的等待线程”
}}}
}
示例2
package lock.demo8;import java.util.concurrent.locks.LockSupport;public class LockSupportTest1 {private static Thread mainThread;public static void main(String[] args) {ThreadA ta &＃61; new ThreadA("ta");// 获取主线程mainThread &＃61; Thread.currentThread();System.out.println(Thread.currentThread().getName() &＃43; " start ta");ta.start();System.out.println(Thread.currentThread().getName() &＃43; " block");// 主线程阻塞
LockSupport.park(mainThread);System.out.println(Thread.currentThread().getName() &＃43; " continue");}static class ThreadA extends Thread {public ThreadA(String name) {super(name);}public void run() {System.out.println(Thread.currentThread().getName() &＃43; " wakup others");// 唤醒“主线程”
LockSupport.unpark(mainThread);}}
}
运行结果&＃xff1a;
main start ta
main block
ta wakup others
main continue
说明&＃xff1a;park和wait的区别。wait让线程阻塞前&＃xff0c;必须通过synchronized获取同步锁。
示例3
LockSupport 很类似于二元信号量(只有1个许可证可供使用)&＃xff0c;如果这个许可还没有被占用&＃xff0c;当前线程获取许可并继 续 执行&＃xff1b;如果许可已经被占用&＃xff0c;当前线 程阻塞&＃xff0c;等待获取许可。
public static void main(String[] args)
{LockSupport.park();System.out.println("block.");
}
运行该代码&＃xff0c;可以发现主线程一直处于阻塞状态。因为 许可默认是被占用的 &＃xff0c;调用park()时获取不到许可&＃xff0c;所以进入阻塞状态。
如下代码&＃xff1a;先释放许可&＃xff0c;再获取许可&＃xff0c;主线程能够正常终止。LockSupport许可的获取和释放&＃xff0c;一般来说是对应的&＃xff0c;如果多次unpark&＃xff0c;只有一次park也不会出现什么问题&＃xff0c;结果是许可处于可用状态。
public static void main(String[] args)
{Thread thread &＃61; Thread.currentThread();LockSupport.unpark(thread);//释放许可LockSupport.park();// 获取许可System.out.println("b");
}
LockSupport是可不重入 的&＃xff0c;如果一个线程连续2次调用 LockSupport.park()&＃xff0c;那么该线程一定会一直阻塞下去。
public static void main(String[] args) throws Exception
{Thread thread &＃61; Thread.currentThread();LockSupport.unpark(thread);System.out.println("a");LockSupport.park();System.out.println("b");LockSupport.park();System.out.println("c");
}
这段代码打印出a和b&＃xff0c;不会打印c&＃xff0c;因为第二次调用park的时候&＃xff0c;线程无法获取许可出现死锁。
下面我们来看下LockSupport对应中断的响应性
public static void t2() throws Exception
{Thread t &＃61; new Thread(new Runnable(){private int count &＃61; 0;&＃64;Overridepublic void run(){long start &＃61; System.currentTimeMillis();long end &＃61; 0;while ((end - start) <&＃61; 1000){count&＃43;&＃43;;end &＃61; System.currentTimeMillis();}System.out.println("after 1 second.count&＃61;" &＃43; count);//等待或许许可
LockSupport.park();System.out.println("thread over." &＃43; Thread.currentThread().isInterrupted());}});t.start();Thread.sleep(2000);// 中断线程
t.interrupt();System.out.println("main over");
}
最终线程会打印出thread over.true。这说明 线程如果因为调用park而阻塞的话&＃xff0c;能够响应中断请求(中断状态被设置成true)&＃xff0c;但是不会抛出InterruptedException 。
转自&＃xff1a;http://blog.csdn.net/aitangyong/article/details/38373137
京公网安备 11010802041100号