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

素颜马尾好姑娘i 2022-08-10 13:53 156阅读 0赞

一、Lock与Synchronized区别

Java中可以使用Lock和Synchronized的可以实现对某个共享资源的同步，同时也可以实现对某些过程的原子性操作。

Lock可以使用Condition进行线程之间的调度，Synchronized则使用Object对象本身的notify, wait, notityAll调度机制，这两种调度机制有什么异同呢？

Condition是Java5以后出现的机制，它有更好的灵活性，而且在一个对象里面可以有多个Condition（即对象监视器),则线程可以注册在不同的Condition，从而可以有选择性的调度线程，更加灵活。

Synchronized就相当于整个对象只有一个单一的Condition（即该对象本身）所有的线程都注册在它身上，线程调度的时候之后调度所有得注册线程，没有选择权，会出现相当大的问题 。

所以，Lock 实现提供了比使用 synchronized 方法和语句可获得的更广泛的锁定操作。此实现允许更灵活的结构，可以具有差别很大的属性，可以支持多个相关的 Condition 对象。

锁是控制多个线程对共享资源进行访问的工具。通常，锁提供了对共享资源的独占访问。一次只能有一个线程获得锁，对共享资源的所有访问都需要首先获得锁。不过，某些锁可能允许对共享资源并发访问，如 ReadWriteLock 的读取锁。

synchronized 方法或语句的使用提供了对与每个对象相关的隐式监视器锁的访问，但却强制所有锁获取和释放均要出现在一个块结构中：当获取了多个锁时，它们必须以相反的顺序释放，且必须在与所有锁被获取时相同的词法范围内释放所有锁。

虽然 synchronized 方法和语句的范围机制使得使用监视器锁编程方便了很多，而且还帮助避免了很多涉及到锁的常见编程错误，但有时也需要以更为灵活的方式使用锁。例如，某些遍历并发访问的数据结果的算法要求使用 "hand-over-hand" 或 "chain locking"：获取节点 A 的锁，然后再获取节点 B 的锁，然后释放 A 并获取 C，然后释放 B 并获取 D，依此类推。Lock 接口的实现允许锁在不同的作用范围内获取和释放，并允许以任何顺序获取和释放多个锁，从而支持使用这种技术。

二、java.util.concurrent.locks类结构

[![locks][]][locks]

上图中，LOCK的实现类其实都是构建在AbstractQueuedSynchronizer上，为何图中没有用UML线表示呢，这是每个Lock实现类都持有自己内部类Sync的实例，而这个Sync就是继承AbstractQueuedSynchronizer(AQS)。为何要实现不同的Sync呢？这和每种Lock用途相关。另外还有AQS的State机制。

基于AQS构建的Synchronizer包括ReentrantLock,Semaphore,CountDownLatch, ReetrantReadWriteLock,FutureTask等，这些Synchronizer实际上最基本的东西就是原子状态的获取和释放，只是条件不一样而已。

ReentrantLock：需要记录当前线程获取原子状态的次数，如果次数为零，那么就说明这个线程放弃了锁（也有可能其他线程占据着锁从而需要等待），如果次数大于1，也就是获得了重进入的效果，而其他线程只能被park住，直到这个线程重进入锁次数变成0而释放原子状态。以下为ReetranLock的FairSync的tryAcquire实现代码解析：

[![image][]][image]

Semaphore：则是要记录当前还有多少次许可可以使用，到0，就需要等待，也就实现并发量的控制，Semaphore一开始设置许可数为1，实际上就是一把互斥锁。以下为Semaphore的FairSync实现：

[![image][image 1]][image 1]

CountDownLatch：闭锁则要保持其状态，在这个状态到达终止态之前，所有线程都会被park住，闭锁可以设定初始值，这个值的含义就是这个闭锁需要被countDown()几次，因为每次CountDown是sync.releaseShared(1),而一开始初始值为10的话，那么这个闭锁需要被countDown()十次，才能够将这个初始值减到0，从而释放原子状态，让等待的所有线程通过。

[![image][image 2]][image 2]

FutureTask：需要记录任务的执行状态，当调用其实例的get方法时,内部类Sync会去调用AQS的acquireSharedInterruptibly()方法，而这个方法会反向调用Sync实现的tryAcquireShared()方法，即让具体实现类决定是否让当前线程继续还是park,而FutureTask的tryAcquireShared方法所做的唯一事情就是检查状态，如果是RUNNING状态那么让当前线程park。而跑任务的线程会在任务结束时调用FutureTask 实例的set方法（与等待线程持相同的实例），设定执行结果，并且通过unpark唤醒正在等待的线程，返回结果。

[![image][image 3]][image 3]

以上4个AQS的使用是比较典型，然而有个问题就是这些状态存在哪里呢？并且是可以计数的。从以上4个example,我们可以很快得到答案，AQS提供给了子类一个int state属性。并且暴露给子类getState()和setState()两个方法(protected)。这样就为上述状态解决了存储问题，RetrantLock可以将这个state用于存储当前线程的重进入次数，Semaphore可以用这个state存储许可数，CountDownLatch则可以存储需要被countDown的次数，而Future则可以存储当前任务的执行状态(RUNING,RAN,CANCELL)。其他的Synchronizer存储他们的一些状态。

AQS留给实现者的方法主要有5个方法，其中tryAcquire,tryRelease和isHeldExclusively三个方法为需要独占形式获取的synchronizer实现的，比如线程独占ReetranLock的Sync，而tryAcquireShared和tryReleasedShared为需要共享形式获取的synchronizer实现。   
ReentrantLock内部Sync类实现的是tryAcquire,tryRelease, isHeldExclusively三个方法(因为获取锁的公平性问题，tryAcquire由继承该Sync类的内部类FairSync和NonfairSync实现)Semaphore内部类Sync则实现了tryAcquireShared和tryReleasedShared(与CountDownLatch相似，因为公平性问题，tryAcquireShared由其内部类FairSync和NonfairSync实现)。CountDownLatch内部类Sync实现了tryAcquireShared和tryReleasedShared。FutureTask内部类Sync也实现了tryAcquireShared和tryReleasedShared。

其实使用过一些JAVA synchronizer的之后，然后结合代码，能够很快理解其到底是如何做到各自的特性的，在把握了基本特性,即获取原子状态和释放原子状态，其实我们自己也可以构造synchronizer。如下是一个LOCK API的一个例子，实现了一个先入先出的互斥锁。

[![image][image 4]][image 4]

三、lock与unlock使用

下面是一个场景，针对这个场景提出两种解决方案。 一个中转站，可以接纳货物，然后发出货物，这是需要建一个仓库，相当于一个缓冲区，当仓库满的时候，不能接货，仓库空的时候，不能发货。

第一种，用一个Condition去解决，有可能会出问题。

> 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();   
>         \}   
>     \}   
> \}

第二种解决方案，用java api中的 一个例子 。

class BoundedBuffer {
       final Lock lock = new ReentrantLock();
       final Condition notFull  = lock.newCondition(); 
       final Condition notEmpty = lock.newCondition(); 
    
       final Object[] items = new Object[100];
       int putptr, takeptr, count;
    
       public void put(Object x) throws InterruptedException {
         lock.lock();
         try {
           while (count == items.length) 
             notFull.await();
           items[putptr] = x; 
           if (++putptr == items.length) putptr = 0;
           ++count;
           notEmpty.signal();
         } finally {
           lock.unlock();
         }
       }
    
       public Object take() throws InterruptedException {
         lock.lock();
         try {
           while (count == 0) 
             notEmpty.await();
           Object x = items[takeptr]; 
           if (++takeptr == items.length) takeptr = 0;
           --count;
           notFull.signal();
           return x;
         } finally {
           lock.unlock();
         }
       } 
     }

[locks]: /images/20220810/0ba4815b3f1c4eb79745670acea1e567.png
[image]: /images/20220810/8439880ac38c4d749081200fa1765fbe.png
[image 1]: /images/20220810/c475f67ec09c4834b75a9eff9f1cc24d.png
[image 2]: /images/20220810/3bcccace457b46869843c330592c9992.png
[image 3]: /images/20220810/916d57eeb22445e7a68ae3eedf9ce368.png
[image 4]: /images/20220810/64829b2b0e284de092ecf48e1223816d.png