Java高级:线程同步lock与unlock使用

mapinfo600 11年前
   <p> </p>    <p>一、Lock与Synchronized区别</p>    <p>Java中可以使用Lock和Synchronized的可以实现对某个共享资源的同步,同时也可以实现对某些过程的原子性操作。</p>    <p>Lock可以使用Condition进行线程之间的调度,Synchronized则使用Object对象本身的notify, wait, notityAll调度机制,这两种调度机制有什么异同呢?</p>    <p>Condition是Java5以后出现的机制,它有更好的灵活性,而且在一个对象里面可以有多个Condition(即对象监视器),则线程可以注册在不同的Condition,从而可以有选择性的调度线程,更加灵活。</p>    <p>Synchronized就相当于整个对象只有一个单一的Condition(即该对象本身)所有的线程都注册在它身上,线程调度的时候之后调度所有得注册线程,没有选择权,会出现相当大的问题 。</p>    <p>所以,Lock 实现提供了比使用 synchronized 方法和语句可获得的更广泛的锁定操作。此实现允许更灵活的结构,可以具有差别很大的属性,可以支持多个相关的 Condition 对象。</p>    <p>锁是控制多个线程对共享资源进行访问的工具。通常,锁提供了对共享资源的独占访问。一次只能有一个线程获得锁,对共享资源的所有访问都需要首先获得锁。不过,某些锁可能允许对共享资源并发访问,如 ReadWriteLock 的读取锁。</p>    <p>synchronized 方法或语句的使用提供了对与每个对象相关的隐式监视器锁的访问,但却强制所有锁获取和释放均要出现在一个块结构中:当获取了多个锁时,它们必须以相反的顺序释放,且必须在与所有锁被获取时相同的词法范围内释放所有锁。</p>    <p>虽然 synchronized 方法和语句的范围机制使得使用监视器锁编程方便了很多,而且还帮助避免了很多涉及到锁的常见编程错误,但有时也需要以更为灵活的方式使用锁。例如,某些遍历并发访问的数据结果的算法要求使用 “hand-over-hand” 或 “chain locking”:获取节点 A 的锁,然后再获取节点 B 的锁,然后释放 A 并获取 C,然后释放 B 并获取 D,依此类推。Lock 接口的实现允许锁在不同的作用范围内获取和释放,并允许以任何顺序获取和释放多个锁,从而支持使用这种技术。</p>    <p>二、java.util.concurrent.locks类结构</p>    <p><img src="https://simg.open-open.com/show/d292bae4deb2d02f71daa8bff0082ae3.jpg"></p>    <p>上图中,LOCK的实现类其实都是构建在AbstractQueuedSynchronizer上,为何图中没有用UML线表示呢,这是每个Lock实现类都持有自己内部类Sync的实例,而这个Sync就是继承AbstractQueuedSynchronizer(AQS)。为何要实现不同的Sync呢?这和每种Lock用途相关。另外还有AQS的State机制。</p>    <p>基于AQS构建的Synchronizer包括 <strong>ReentrantLock</strong> , <strong>Semaphore</strong> , <strong>CountDownLatch</strong> ,  <strong>ReetrantRead</strong> <strong>WriteLock</strong> , <strong>FutureTask</strong> 等,这些Synchronizer实际上最基本的东西就是原子状态的获取和释放,只是条件不一样而已。</p>    <p>ReentrantLock:需要记录当前线程获取原子状态的次数,如果次数为零,那么就说明这个线程放弃了锁(也有可能其他线程占据着锁从而需要等待),如果次数大于1,也就是获得了重进入的效果,而其他线程只能被park住,直到这个线程重进入锁次数变成0而释放原子状态。以下为ReetranLock的FairSync的tryAcquire实现代码解析:</p>    <p><img src="https://simg.open-open.com/show/c4f0a9149e043b787193fa14eb5c4725.png"></p>    <p>Semaphore:则是要记录当前还有多少次许可可以使用,到0,就需要等待,也就实现并发量的控制,Semaphore一开始设置许可数为1,实际上就是一把互斥锁。以下为Semaphore的FairSync实现:</p>    <p><img src="https://simg.open-open.com/show/51cd139abbadcb389f38954db4105563.png"></p>    <p>CountDownLatch:闭锁则要保持其状态,在这个状态到达终止态之前,所有线程都会被park住,闭锁可以设定初始值,这个值的含义就是这个闭锁需要被countDown()几次,因为每次CountDown是sync.releaseShared(1),而一开始初始值为10的话,那么这个闭锁需要被countDown()十次,才能够将这个初始值减到0,从而释放原子状态,让等待的所有线程通过。</p>    <p><img src="https://simg.open-open.com/show/768d917ee9374fdc54651b367b59e7a5.png"></p>    <p>FutureTask:需要记录任务的执行状态,当调用其实例的get方法时,内部类Sync会去调用AQS的acquireSharedInterruptibly()方法,而这个方法会反向调用Sync实现的tryAcquireShared()方法,即让具体实现类决定是否让当前线程继续还是park,而FutureTask的tryAcquireShared方法所做的唯一事情就是检查状态,如果是RUNNING状态那么让当前线程park。而跑任务的线程会在任务结束时调用FutureTask 实例的set方法(与等待线程持相同的实例),设定执行结果,并且通过unpark唤醒正在等待的线程,返回结果。</p>    <p><img src="https://simg.open-open.com/show/f627373d94b834b1ac8f11e03e9bb04b.png"></p>    <p>以上4个AQS的使用是比较典型,然而有个问题就是这些状态存在哪里呢?并且是可以计数的。从以上4个example,我们可以很快得到答案,AQS提供给了子类一个int state属性。并且暴露给子类getState()和setState()两个方法(protected)。这样就为上述状态解决了存储问题,RetrantLock可以将这个state用于存储当前线程的重进入次数,Semaphore可以用这个state存储许可数,CountDownLatch则可以存储需要被countDown的次数,而Future则可以存储当前任务的执行状态(RUNING,RAN,CANCELL)。其他的Synchronizer存储他们的一些状态。</p>    <p>AQS留给实现者的方法主要有5个方法,其中tryAcquire,tryRelease和isHeldExclusively三个方法为需要独占形式获取的synchronizer实现的,比如线程独占ReetranLock的Sync,而tryAcquireShared和tryReleasedShared为需要共享形式获取的synchronizer实现。</p>    <p>ReentrantLock内部Sync类实现的是tryAcquire,tryRelease, isHeldExclusively三个方法(因为获取锁的公平性问题,tryAcquire由继承该Sync类的内部类FairSync和NonfairSync实现) <strong>Semaphore</strong> 内部类Sync则实现了tryAcquireShared和tryReleasedShared(与CountDownLatch相似,因为公平性问题,tryAcquireShared由其内部类FairSync和NonfairSync实现)。 <strong>CountDownLatch</strong> 内部类Sync实现了tryAcquireShared和tryReleasedShared。 <strong>FutureTask</strong> 内部类Sync也实现了tryAcquireShared和tryReleasedShared。</p>    <p>其实使用过一些JAVA synchronizer的之后,然后结合代码,能够很快理解其到底是如何做到各自的特性的,在把握了基本特性,即获取原子状态和释放原子状态,其实我们自己也可以构造synchronizer。如下是一个LOCK API的一个例子,实现了一个先入先出的互斥锁。</p>    <p><img src="https://simg.open-open.com/show/3a2b494b773cdc2375ad61ac68bb678b.png"></p>    <p>三、lock与unlock使用</p>    <p>下面是一个场景,针对这个场景提出两种解决方案。 一个中转站,可以接纳货物,然后发出货物,这是需要建一个仓库,相当于一个缓冲区,当仓库满的时候,不能接货,仓库空的时候,不能发货。</p>    <p>第一种,用一个Condition去解决,有可能会出问题。</p>    <pre>  <code class="language-java">package com.zxx;    import java.util.Random;   import java.util.concurrent.ExecutorService;   import java.util.concurrent.Executors;   import java.util.concurrent.locks.Condition;   import java.util.concurrent.locks.Lock;   import java.util.concurrent.locks.ReentrantLock;    /**   * 单个Condition去控制一个缓冲区,多线程对缓冲区做读写操作,要保证缓冲区满的时侯不会   * 被写,空的时候不会被读;单个Condition控制会出错误: 当缓冲区还有一个位置时,多个写线程   * 同时访问,则只有一个写线程可以对其进行写操作,操作完之后,唤醒在这个condition上等待的   * 其他几个写线程,如果判断用IF语句的话就会出现继续向缓冲区添加。   * @author Administrator   *   */   public class ConditionError {       Lock lock = new ReentrantLock();       Condition condition = lock.newCondition();       String[] container = new String[10];       int index = 0;       public static void main(String[] args) {           ConditionError conditionError = new ConditionError();           conditionError.test();       }       public void test(){           ExecutorService threadPool = Executors.newCachedThreadPool();           for(int i = 0; i < 14; i++){//先用14个线程去写,则有4个线程会被阻塞               threadPool.execute(new Runnable(){                    @Override                   public void run() {                       put();                   }               });           }           Executors.newSingleThreadExecutor().execute(new Runnable(){//用一个线程去取,则会通知4个阻塞的写线程工作,此时                                                                   //会有一个线程向缓冲区写,写完后去通知在这个condition上等待                                                                   //的取线程,这是它的本意,但是它唤醒了写线程,因为只有一个condition                                                                   //不能有选择的唤醒写取线程,此时就需要有多个Condition               @Override               public void run() {                   try {                       Thread.sleep(10000);                   } catch (InterruptedException e) {                       e.printStackTrace();                   }                   get();               }           });       }       /**        * 向缓冲去写数据        */       public void put(){           lock.lock();           try{               System.out.println(Thread.currentThread().getName() + "当前位置:" + index + "-----------------------------");               while(index == 10){                   try {                       System.out.println(Thread.currentThread().getName() + "处于阻塞状态!");                       condition.await();   //                    index = 0;                   } catch (InterruptedException e) {                       e.printStackTrace();                   }               }               container[index] = new String(new Random().nextInt() + "");               condition.signalAll();               index ++;           } finally {               lock.unlock();           }       }       /**        * 从缓冲区拿数据        */       public void get(){           lock.lock();           try{               while(index == 0){                   try {                       System.out.println("get--------" + Thread.currentThread().getName() + "处于阻塞");                       condition.await();                   } catch (InterruptedException e) {                       e.printStackTrace();                   }               }               index --;               System.out.println("get---------" + Thread.currentThread().getName() + "唤醒阻塞");               condition.signalAll();           } finally {               lock.unlock();           }       }   }</code></pre>    <p>第二种解决方案,用java api中的 一个例子 。</p>    <p><img src="https://simg.open-open.com/show/93f8fff269ac0efd3f5501705177cc0d.png"></p>    <p>来源:<a href="/misc/goto?guid=4959671523424891727">xianggao</a></p>