zoukankan      html  css  js  c++  java
  • Java多线程系列--AQS之 LockSupport

    concurrent包是基于AQS (AbstractQueuedSynchronizer)框架的,AQS(JAVA CAS原理、unsafe、AQS)框架借助于两个类:

     因此,LockSupport非常重要。

     LockSupport介绍

      LockSupport是JDK中比较底层的类,用来创建锁和其他同步工具类的基本线程阻塞原语。java锁和同步器框架的核心 AQS: AbstractQueuedSynchronizer,就是通过调用 LockSupport .park()和 LockSupport .unpark()实现线程的阻塞和解除阻塞的。LockSupport中的park() 和 unpark() 的作用分别是阻塞线程和解除阻塞线程,而且park()和unpark()不会遇到“Thread.suspend 和 Thread.resume所可能引发的死锁”问题

    LockSupport类是Java6(JSR166-JUC)引入的一个类,提供了基本的线程同步原语。LockSupport实际上是调用了Unsafe类里的函数,归结到Unsafe里,只有两个函数:

    •  park:阻塞当前线程(Block current thread),字面理解park,就算占住,停车的时候不就把这个车位给占住了么?起这个名字还是很形象的。
    • unpark: 使给定的线程停止阻塞(Unblock the given thread blocked )。

    因为park() 和 unpark()有许可的存在;调用 park() 的线程和另一个试图将其 unpark() 的线程之间的竞争将保持活性。

     类图:

    LockSupport函数列表

    //
    private static void setBlocker(Thread t, Object arg)
    // 返回提供给最近一次尚未解除阻塞的 park 方法调用的 blocker 对象,如果该调用不受阻塞,则返回 null。
    static Object getBlocker(Thread t)
    // 为了线程调度,禁用当前线程,除非许可可用。
    static void park()
    // 为了线程调度,在许可可用之前禁用当前线程。
    static void park(Object blocker)
    // 为了线程调度禁用当前线程,最多等待指定的等待时间,除非许可可用。
    static void parkNanos(long nanos)
    // 为了线程调度,在许可可用前禁用当前线程,并最多等待指定的等待时间。
    static void parkNanos(Object blocker, long nanos)
    // 为了线程调度,在指定的时限前禁用当前线程,除非许可可用。
    static void parkUntil(long deadline)
    // 为了线程调度,在指定的时限前禁用当前线程,除非许可可用。
    static void parkUntil(Object blocker, long deadline)
    // 如果给定线程的许可尚不可用,则使其可用。
    static void unpark(Thread thread)

    两个重点

    (1)操作对象

    归根结底,LockSupport调用的Unsafe中的native代码: 

        public static void unpark(Thread thread) {
            if (thread != null)
                UNSAFE.unpark(thread);
        }
    
        public static void park() {
            UNSAFE.park(false, 0L);
        }

    两个函数声明清楚地说明了操作对象:park函数是将当前Thread阻塞,而unpark函数则是将指定线程Thread唤醒

    与Object类的wait/notify机制相比,park/unpark有两个优点:

    1. 以thread为操作对象更符合阻塞线程的直观定义;

    2. 操作更精准,可以准确地唤醒某一个线程(notify随机唤醒一个线程,notifyAll唤醒所有等待的线程),增加了灵活性。

    (2)关于许可

    在上面的文字中,我使用了阻塞和唤醒,是为了和wait/notify做对比。

    • 其实park/unpark的设计原理核心是“许可”。park是等待一个许可。unpark是为某线程提供一个许可。如果某线程A调用park,那么除非另外一个线程调用unpark(A)给A一个许可,否则线程A将阻塞在park操作上。
    • 有一点比较难理解的,是unpark操作可以再park操作之前。也就是说,先提供许可。当某线程调用park时,已经有许可了,它就消费这个许可,然后可以继续运行。这其实是必须的。考虑最简单的生产者(Producer)消费者(Consumer)模型:Consumer需要消费一个资源,于是调用park操作等待;Producer则生产资源,然后调用unpark给予Consumer使用的许可。非常有可能的一种情况是,Producer先生产,这时候Consumer可能还没有构造好(比如线程还没启动,或者还没切换到该线程)。那么等Consumer准备好要消费时,显然这时候资源已经生产好了,可以直接用,那么park操作当然可以直接运行下去。如果没有这个语义,那将非常难以操作。
    • 但是这个“许可”是不能叠加的,“许可”是一次性的。比如线程B连续调用了三次unpark函数,当线程A调用park函数就使用掉这个“许可”,如果线程A再次调用park,则进入等待状态。

    park和unpark的灵活之处

    上面已经提到,unpark函数可以先于park调用,这个正是它们的灵活之处。

    一个线程它有可能在别的线程unPark之前,或者之后,或者同时调用了park,那么因为park的特性,它可以不用担心自己的park的时序问题,否则,如果park必须要在unpark之前,那么给编程带来很大的麻烦!!

    考虑一下,两个线程同步,要如何处理?

    在Java5里是用wait/notify/notifyAll来同步的。wait/notify机制有个很蛋疼的地方是,比如线程B要用notify通知线程A,那么线程B要确保线程A已经在wait调用上等待了,否则线程A可能永远都在等待。编程的时候就会很蛋疼。

    另外,是调用notify,还是notifyAll?

    notify只会唤醒一个线程,如果错误地有两个线程在同一个对象上wait等待,那么又悲剧了。为了安全起见,貌似只能调用notifyAll了。而unpark可以指定到特定线程。

    park/unpark模型真正解耦了线程之间的同步,线程之间不再需要一个Object或者其它变量来存储状态,不再需要关心对方的状态。

    Unsafe.park和Unsafe.unpark的底层实现原理

    在Linux系统下,是用的Posix线程库pthread中的mutex(互斥量),condition(条件变量)来实现的。
    mutex和condition保护了一个_counter的变量,当park时,这个变量被设置为0,当unpark时,这个变量被设置为1。

    源码:
    每个Java线程都有一个Parker实例,Parker类是这样定义的:
    class Parker : public os::PlatformParker {  
    private:  
      volatile int _counter ;  
      ...  
    public:  
      void park(bool isAbsolute, jlong time);  
      void unpark();  
      ...  
    }  
    class PlatformParker : public CHeapObj<mtInternal> {  
      protected:  
        pthread_mutex_t _mutex [1] ;  
        pthread_cond_t  _cond  [1] ;  
        ...  
    }  

     可以看到Parker类实际上用Posix的mutex,condition来实现的。
    在Parker类里的_counter字段,就是用来记录“许可”的。

    • park 过程

    当调用park时,先尝试能否直接拿到“许可”,即_counter>0时,如果成功,则把_counter设置为0,并返回:

    void Parker::park(bool isAbsolute, jlong time) {  
      
      // Ideally we'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;  

    如果不成功,则构造一个ThreadBlockInVM,然后检查_counter是不是>0,如果是,则把_counter设置为0,unlock mutex并返回:

    ThreadBlockInVM tbivm(jt);  
    if (_counter > 0)  { // no wait needed  
      _counter = 0;  
      status = pthread_mutex_unlock(_mutex);  

    否则,再判断等待的时间,然后再调用pthread_cond_wait函数等待,如果等待返回,则把_counter设置为0,unlock mutex并返回:

    if (time == 0) {  
      status = pthread_cond_wait (_cond, _mutex) ;  
    }  
    _counter = 0 ;  
    status = pthread_mutex_unlock(_mutex) ;  
    assert_status(status == 0, status, "invariant") ;  
    OrderAccess::fence();  
    • unpark 过程

    当unpark时,则简单多了,直接设置_counter为1,再unlock mutex返回。如果_counter之前的值是0,则还要调用pthread_cond_signal唤醒在park中等待的线程:

    void Parker::unpark() {  
      int s, status ;  
      status = pthread_mutex_lock(_mutex);  
      assert (status == 0, "invariant") ;  
      s = _counter;  
      _counter = 1;  
      if (s < 1) {  
         if (WorkAroundNPTLTimedWaitHang) {  
            status = pthread_cond_signal (_cond) ;  
            assert (status == 0, "invariant") ;  
            status = pthread_mutex_unlock(_mutex);  
            assert (status == 0, "invariant") ;  
         } else {  
            status = pthread_mutex_unlock(_mutex);  
            assert (status == 0, "invariant") ;  
            status = pthread_cond_signal (_cond) ;  
            assert (status == 0, "invariant") ;  
         }  
      } else {  
        pthread_mutex_unlock(_mutex);  
        assert (status == 0, "invariant") ;  
      }  
    }  

    LockSupport示例

    对比下面的“示例1”和“示例2”可以更清晰的了解LockSupport的用法。

    示例1

    package lock.demo7;
    
    public class WaitTest1 {
    
        public static void main(String[] args) {
    
            ThreadA ta = new ThreadA("ta");
    
            synchronized (ta) { // 通过synchronized(ta)获取“对象ta的同步锁”
                try {
                    System.out.println(Thread.currentThread().getName() + " start ta");
                    ta.start();
    
                    System.out.println(Thread.currentThread().getName() + " block");
                    // 主线程等待
                    ta.wait();
    
                    System.out.println(Thread.currentThread().getName() + " 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() + " 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 = new ThreadA("ta");
            // 获取主线程
            mainThread = Thread.currentThread();
    
            System.out.println(Thread.currentThread().getName() + " start ta");
            ta.start();
    
            System.out.println(Thread.currentThread().getName() + " block");
            // 主线程阻塞
            LockSupport.park(mainThread);
    
            System.out.println(Thread.currentThread().getName() + " continue");
        }
    
        static class ThreadA extends Thread {
    
            public ThreadA(String name) {
                super(name);
            }
    
            public void run() {
                System.out.println(Thread.currentThread().getName() + " wakup others");
                // 唤醒“主线程”
                LockSupport.unpark(mainThread);
            }
        }
    }

    运行结果

    main start ta
    main block
    ta wakup others
    main continue

    说明park和wait的区别。wait让线程阻塞前,必须通过synchronized获取同步锁。

    示例3

    LockSupport 很类似于二元信号量(只有1个许可证可供使用),如果这个许可还没有被占用,当前线程获取许可并继 续 执行;如果许可已经被占用,当前线 程阻塞,等待获取许可。

    public static void main(String[] args)
    {
         LockSupport.park();
         System.out.println("block.");
    }

    运行该代码,可以发现主线程一直处于阻塞状态。因为 许可默认是被占用的 ,调用park()时获取不到许可,所以进入阻塞状态。

    如下代码:先释放许可,再获取许可,主线程能够正常终止。LockSupport许可的获取和释放,一般来说是对应的,如果多次unpark,只有一次park也不会出现什么问题,结果是许可处于可用状态。

    public static void main(String[] args)
    {
         Thread thread = Thread.currentThread();
         LockSupport.unpark(thread);//释放许可
         LockSupport.park();// 获取许可
         System.out.println("b");
    }

    LockSupport是可不重入 的,如果一个线程连续2次调用 LockSupport.park(),那么该线程一定会一直阻塞下去。

    public static void main(String[] args) throws Exception
    {
        Thread thread = Thread.currentThread();
        
        LockSupport.unpark(thread);
        
        System.out.println("a");
        LockSupport.park();
        System.out.println("b");
        LockSupport.park();
        System.out.println("c");
    }

    这段代码打印出a和b,不会打印c,因为第二次调用park的时候线程无法获取许可出现死锁

    下面我们来看下LockSupport对应中断的响应性

    public static void t2() throws Exception
    {
        Thread t = new Thread(new Runnable()
        {
            private int count = 0;
    
            @Override
            public void run()
            {
                long start = System.currentTimeMillis();
                long end = 0;
    
                while ((end - start) <= 1000)
                {
                    count++;
                    end = System.currentTimeMillis();
                }
    
                System.out.println("after 1 second.count=" + count);
    
                //等待或许许可
                LockSupport.park();
                System.out.println("thread over." + Thread.currentThread().isInterrupted());
    
            }
        });
    
        t.start();
    
        Thread.sleep(2000);
    
        // 中断线程
        t.interrupt();
    
        
        System.out.println("main over");
    }

    最终线程会打印出thread over.true。这说明 线程如果因为调用park而阻塞的话,能够响应中断请求(中断状态被设置成true),但是不会抛出InterruptedException 。

    转自:http://blog.csdn.net/aitangyong/article/details/38373137

  • 相关阅读:
    函数、包和错误处理
    程序流程控制
    poj 2515 Birthday Cake
    poj 2094 多项式求和。
    hdu 3625 第一类striling 数
    hdu 4372 第一类stirling数的应用/。。。好题
    poj 1845 Sumdiv
    hdu 3641 Treasure Hunting 强大的二分
    poj 3335 /poj 3130/ poj 1474 半平面交 判断核是否存在 / poj1279 半平面交 求核的面积
    hdu 2841 Visible Trees
  • 原文地址:https://www.cnblogs.com/duanxz/p/6063699.html
Copyright © 2011-2022 走看看