[JDK] 多线程高并发探秘之“锁”

情人节快乐!

[JDK] 多线程高并发探秘之“锁”

锁作为并发共享数据,保证一致性的工具,在JAVA平台有多种实现(如 synchronized 和 ReentrantLock等等 ) 。这些已经写好提供的锁为我们开发提供了便利,但是锁的具体性质以及类型却很少被提及。

1. 自旋锁

自旋锁是采用让当前线程不停地的在循环体内执行实现的,当循环的条件被其他线程改变时 才能进入临界区。

Ex1

@RequestMapping(value = "lock/{index}", method = RequestMethod.GET)public void lock(@PathVariable Integer index) {   switch (index) {       case 0:
          concurrenceLock.doTest();           break;       case 1:
          concurrenceLock.doTest1();           break;       case 2:
          concurrenceLock.doTest2();           break;       default:           break;
  }
}
package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;import java.util.Objects;import java.util.concurrent.atomic.AtomicReference;/**
* <p>
* 自旋锁
* 自旋锁是采用让当前线程不停地的在循环体内执行实现的,当循环的条件被其他线程改变时 才能进入临界区。
* 自旋锁是指尝试获取锁的线程不会立即阻塞,而是采用循环的方式去尝试获取锁,这样的好处是减少线程上下文切换的消耗,缺点是循环会消耗CPU。
* </p>
*
* @author xiachaoyang
* @version V1.0
* @date 2019年01月23日 15:33
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年01月23日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class SpinLock {    private AtomicReference<Thread> sign = new AtomicReference<>();    public void lock(){
       Thread current = Thread.currentThread();        //如果当前值{@code ==}为期望值,原子地将值设置为给定的更新值
       log.debug("{} spinLock start lock..........",  current.getName());        while (!sign.compareAndSet(null, current)) {
           log.debug("{} spinLock locking........", current.getName());
       }
       log.debug("{} spinLock quit lock..........",  current.getName());
   }    public void unlock() {
       Thread current = Thread.currentThread();        boolean flag = sign.compareAndSet(current, null);
       log.debug("{} spinLock unlock() >>>> {}",current.getName(),flag);
   }
}//ConcurrenceLockServiceImplpackage com.example.service.concurrence.impl;import com.example.concurrence.lock.SpinLock;import com.example.service.concurrence.ConcurrenceService;import lombok.extern.slf4j.Slf4j;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import org.springframework.stereotype.Service;import javax.annotation.Resource;/**
* <p>
*
* </p>
*
* @author xiachaoyang
* @version V1.0
* @date 2019年01月23日 15:42
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年01月23日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4j@Servicepublic class ConcurrenceLockServiceImpl implements ConcurrenceService.LockPart {    private static SpinLock spinLock = new SpinLock();    private Thread last;    @Resource
   private ThreadPoolTaskExecutor threadPoolTaskExecutor;     @Override
   public void doTest() throws InterruptedException {
       spinLock.lock();
       Thread.sleep(10000);
       spinLock.unlock();
   }
}

每次执行完自旋的判断后,sign的引用会被指向当前线程,下次进入判断后,预测值和当前线程不一致,则会返回false,即进入自旋锁循环体内 sign.compareAndSet(null, current)该方法,前者为预测原来AtomicReference的引用值,后者为更新值,预测正确的话,则更新为更新值,返回true;预测错误则返回false。


使用了CAS原子操作,lock函数将owner设置为当前线程,并且预测原来的值为空。unlock函数将owner设置为null,并且预测值为当前线程。

  1. 当有第二个线程调用lock操作时由于owner值不为空,导致循环一直被执行,直至第一个线程调用unlock函数将owner设置为null,第二个线程才能进入临界区。
  2. 由于自旋锁只是将当前线程不停地执行循环体,不进行线程状态的改变,所以响应速度更快。但当线程数不停增加时,性能下降明显,因为每个线程都需要执行,占用CPU时间。如果线程竞争不激烈,并且保持锁的时间段。适合使用自旋锁。

注:该例子为非公平锁,获得锁的先后顺序,不会按照进入lock的先后顺序进行。

日志

//单次调用
2019-01-28 17:40:38.553 DEBUG 8496 --- [nio-8080-exec-2] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-2 spinLock start lock..........
2019-01-28 17:40:38.553 DEBUG 8496 --- [nio-8080-exec-2] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-2 spinLock quit lock..........
2019-01-28 17:40:48.553 DEBUG 8496 --- [nio-8080-exec-2] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-2 spinLock unlock() >>>> true
//并发调用
2019-01-28 17:44:07.431 DEBUG 8496 --- [nio-8080-exec-8] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-8 spinLock locking........
2019-01-28 17:44:07.432 DEBUG 8496 --- [nio-8080-exec-8] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-8 spinLock quit lock..........
2019-01-28 17:44:07.432 DEBUG 8496 --- [nio-8080-exec-6] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-6 spinLock unlock() >>>> true
2019-01-28 17:44:17.432 DEBUG 8496 --- [nio-8080-exec-8] com.example.concurrence.lock.SpinLock    : http-nio-8080-exec-8 spinLock unlock() >>>> true

2. 自旋锁的其他种类

除了前文提到的自旋锁,在自旋锁中另有三种常见的锁形式:TicketLock,CLHlockMCSlock

2.1 TicketSpinLock

package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.atomic.AtomicInteger;/**
* <p>
* Ticket锁主要解决的是访问顺序的问题,主要的问题是在多核cpu上
* 但是每次都要查询一个serviceNum 服务号,影响性能(必须要到主内存读取,并阻止其他cpu修改)
* </p>
*
* @author xiachaoyang
* @version V1.0
* @date 2019年01月24日 10:58
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年01月24日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class TicketSpinLock {    private AtomicInteger serviceNum = new AtomicInteger();    private AtomicInteger ticketNum = new AtomicInteger();    private static final ThreadLocal<Integer> LOCAL = new ThreadLocal<Integer>();    public void lock() {
       log.debug("{} try lock..................",Thread.currentThread().getName());        int myticket = ticketNum.getAndIncrement();
       LOCAL.set(myticket);
       log.debug("{} set in LOCAL..................",myticket);        while (myticket != serviceNum.get()) {
           log.debug("{} locking..................",Thread.currentThread().getName());
       }
   }    public void unlock() {        int myticket = LOCAL.get();        boolean flag = serviceNum.compareAndSet(myticket, myticket + 1);
       log.debug("{} unlocked..................>>myticket is {}, and flag is {}",Thread.currentThread().getName(),myticket,flag);
   }    //浅析TicketLock(https://blog.csdn.net/yxc5463/article/details/78193991)
   //Java多线程编程排队锁(Ticket Lock详解)(http://www.leftso.com/blog/466.html)}

测试代码

@Overridepublic void doTest3() throws InterruptedException {
   ticketSpinLock.lock();
   Thread.sleep(10000);
   ticketSpinLock.unlock();
}//模拟多线程,利用PostMan触发请求,调用接口;中间停顿10s是为了模拟线程阻塞

日志

2019-01-28 16:43:30.339 DEBUG 13280 --- [nio-8080-exec-9] c.e.concurrence.lock.TicketSpinLock      : http-nio-8080-exec-9 locking..................
2019-01-28 16:43:30.339 DEBUG 13280 --- [nio-8080-exec-9] c.e.concurrence.lock.TicketSpinLock      : http-nio-8080-exec-9 locking..................
2019-01-28 16:43:30.339 DEBUG 13280 --- [nio-8080-exec-9] c.e.concurrence.lock.TicketSpinLock      : http-nio-8080-exec-9 locking..................
2019-01-28 16:43:30.561 DEBUG 13280 --- [nio-8080-exec-7] c.e.concurrence.lock.TicketSpinLock      : http-nio-8080-exec-7 unlocked..................>>myticket is 3, and flag is true
2019-01-28 16:43:31.108 DEBUG 13280 --- [nio-8080-exec-7] o.s.b.w.s.f.OrderedRequestContextFilter  : Cleared thread-bound request context: org.apache.catalina.connector.RequestFacade@b5f05b5
2019-01-28 16:43:41.108 DEBUG 13280 --- [nio-8080-exec-9] c.e.concurrence.lock.TicketSpinLock      : http-nio-8080-exec-9 unlocked..................>>myticket is 4, and flag is true

缺点

Ticket Lock 虽然解决了公平性的问题,但是多处理器系统上,每个进程/线程占用的处理器都在读写同一个变量serviceNum ,每次读写操作都必须在多个处理器缓存之间进行缓存同步,这会导致繁重的系统总线和内存的流量,大大降低系统整体的性能。

2.2 CLHLock & MCSLock

CLHLock 和MCSLock 则是两种类型相似的公平锁,采用链表的形式进行排序

CLHLock

package com.example.concurrence.lock;import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;/**
* <p>
*  CLHlock是不停的查询前驱变量, 导致不适合在NUMA 架构下使用(在这种结构下,每个线程分布在不同的物理内存区域)
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年01月28日 17:04
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年01月28日
* @modify reason: {方法名}:{原因}
* ...
*/public class CLHLock {    public static class CLHNode {        private volatile boolean isLocked = true;
   }    @SuppressWarnings("unused")    private volatile CLHNode                                           tail;    private static final ThreadLocal<CLHNode>                          LOCAL   = new ThreadLocal<CLHNode>();    private static final AtomicReferenceFieldUpdater<CLHLock, CLHNode> UPDATER = AtomicReferenceFieldUpdater.newUpdater(CLHLock.class,
           CLHNode.class, "tail");    public void lock() {
       CLHNode node = new CLHNode();
       LOCAL.set(node);
       CLHNode preNode = UPDATER.getAndSet(this, node);        if (preNode != null) {            while (preNode.isLocked) {
           }
           preNode = null;
           LOCAL.set(node);
       }
   }    public void unlock() {
       CLHNode node = LOCAL.get();        if (!UPDATER.compareAndSet(this, node, null)) {
           node.isLocked = false;
       }
       node = null;
   }
}

MCSLock

package com.example.concurrence.lock;import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;/**
* <p>
* MCSLock则是对本地变量的节点进行循环,不存在CLHlock 的问题。
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年01月28日 17:11
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年01月28日
* @modify reason: {方法名}:{原因}
* ...
*/public class MCSLock {    public static class MCSNode {        volatile MCSNode next;        volatile boolean isLocked = true;
   }    private static final ThreadLocal<MCSNode>                          NODE    = new ThreadLocal<MCSNode>();    @SuppressWarnings("unused")    private volatile MCSNode                                           queue;    private static final AtomicReferenceFieldUpdater<MCSLock, MCSNode> UPDATER = AtomicReferenceFieldUpdater.newUpdater(MCSLock.class,
           MCSNode.class, "queue");    public void lock() {
       MCSNode currentNode = new MCSNode();
       NODE.set(currentNode);
       MCSNode preNode = UPDATER.getAndSet(this, currentNode);        if (preNode != null) {
           preNode.next = currentNode;            while (currentNode.isLocked) {           }
       }
   }    public void unlock() {
       MCSNode currentNode = NODE.get();        if (currentNode.next == null) {            if (UPDATER.compareAndSet(this, currentNode, null)) {           } else {                while (currentNode.next == null) {
               }                // 释放锁
               currentNode.next.isLocked = false;
               currentNode.next = null;
           }
       } else {
           currentNode.next.isLocked = false;
           currentNode.next = null;
       }
   }
}

CAS

/**AtomicReferenceFieldUpdater*/public final boolean compareAndSet(T obj, V expect, V update) {
   accessCheck(obj);
   valueCheck(update);    /**this.offset = U.objectFieldOffset(field);初始化构造时写入   public static <U,W> AtomicReferenceFieldUpdater<U,W> newUpdater(Class<U> tclass,Class<W> vclass,String fieldName) {
       return new AtomicReferenceFieldUpdaterImpl<U,W>
           (tclass, vclass, fieldName, Reflection.getCallerClass());
   }   AtomicReferenceFieldUpdaterImpl(final Class<T> tclass,
                                       final Class<V> vclass,
                                       final String fieldName,
                                       final Class<?> caller)
   */
   return U.compareAndSwapObject(obj, offset, expect, update);
}/**
* 比较obj的offset处内存位置中的值和期望的值,如果相同则更新。此更新是不可中断的。
*
* @param obj 需要更新的对象
* @param offset obj中整型field的偏移量
* @param expect 希望field中存在的值
* @param update 如果期望值expect与field的当前值相同,设置filed的值为这个新值
* @return 如果field的值被更改返回true
*/public native boolean compareAndSwapInt(Object obj, long offset, int expect, int update);

CAS操作有3个操作数,内存值M,预期值E,新值U,如果M==E,则将内存值修改为B,否则啥都不做。

Java中具体的CAS操作类sun.misc.Unsafe。Unsafe类提供了硬件级别的原子操作,Java无法直接访问到操作系统底层(如系统硬件等),为此Java使用native方法来扩展Java程序的功能。

日志

2019-02-01 11:47:22.539 DEBUG 8948 --- [nio-8080-exec-1] com.example.concurrence.lock.MCSLock     : http-nio-8080-exec-1 MCSLock unlock()
2019-02-01 11:47:22.539 DEBUG 8948 --- [nio-8080-exec-5] com.example.concurrence.lock.MCSLock     : http-nio-8080-exec-5 MCSLock quit lock..........
2019-02-01 11:47:22.564 DEBUG 8948 --- [nio-8080-exec-1] o.s.b.w.s.f.OrderedRequestContextFilter  : Cleared thread-bound request context: org.apache.catalina.connector.RequestFacade@27711663
2019-02-01 11:47:32.540 DEBUG 8948 --- [nio-8080-exec-5] com.example.concurrence.lock.MCSLock     : http-nio-8080-exec-5 MCSLock unlock()

分析

  • 从代码上 看,CLH 要比 MCS 更简单;
  • CLH 的队列是隐式的队列,没有真实的后继结点属性。
  • MCS 的队列是显式的队列,有真实的后继结点属性。

JUC ReentrantLock 默认内部使用的锁 即是 CLH锁(有很多改进的地方,将自旋锁换成了阻塞锁等等)。

3.阻塞锁

阻塞锁,与自旋锁不同,改变了线程的运行状态。在JAVA环境中,线程Thread有如下几个状态:

  • 新建状态
  • 就绪状态
  • 运行状态
  • 阻塞状态
  • 死亡状态

阻塞锁,可以说是让线程进入阻塞状态进行等待,当获得相应的信号(唤醒,时间) 时,才可以进入线程的准备就绪状态,准备就绪状态的所有线程,通过竞争,进入运行状态。 JAVA中,能够进入\退出、阻塞状态或包含阻塞锁的方法有 ,synchronized 关键字(其中的重量锁),ReentrantLockObject.wait()\notify(),LockSupport.park()/unpart()(j.u.c经常使用)

package com.example.concurrence.lock;import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;import java.util.concurrent.locks.LockSupport;/**
* <p>
*  阻塞锁,可以说是让线程进入阻塞状态进行等待,当获得相应的信号(唤醒,时间) 时,才可以进入线程的准备就绪状态,准备就绪状态的所有线程,通过竞争,进入运行状态。
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年02月13日 10:00
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年02月13日
* @modify reason: {方法名}:{原因}
* ...
*/public class CLHLockWithBlock {    public static class CLHNode {        private volatile Thread isLocked;
   }    @SuppressWarnings("unused")    private volatile CLHNode                                            tail;    private static final ThreadLocal<CLHNode>                           LOCAL   = new ThreadLocal<CLHNode>();    private static final AtomicReferenceFieldUpdater<CLHLockWithBlock, CLHNode> UPDATER = AtomicReferenceFieldUpdater.newUpdater(CLHLockWithBlock.class,
           CLHNode.class, "tail");    public void lock() {
       CLHNode node = new CLHNode();
       LOCAL.set(node);
       CLHNode preNode = UPDATER.getAndSet(this, node);        if (preNode != null) {
           preNode.isLocked = Thread.currentThread();
           LockSupport.park(this);
           preNode = null;
           LOCAL.set(node);
       }
   }    public void unlock() {
       CLHNode node = LOCAL.get();        if (!UPDATER.compareAndSet(this, node, null)) {
           System.out.println("unlock\t" + node.isLocked.getName());
           LockSupport.unpark(node.isLocked);
       }
       node = null;
   }
}

在这里我们使用了LockSupport.unpark()的阻塞锁。 该例子是将CLH锁修改而成。

阻塞锁的优势在于,阻塞的线程不会占用CPU时间, 不会导致 CPU占用率过高,但进入时间以及恢复时间都要比自旋锁略慢。

在竞争激烈的情况下 阻塞锁的性能要明显高于自旋锁。

理想的情况则是; 在线程竞争不激烈的情况下,使用自旋锁,竞争激烈的情况下使用阻塞锁

LockSupport

    public static void park(Object blocker) {
       Thread t = Thread.currentThread();
       setBlocker(t, blocker);
       UNSAFE.park(false, 0L);
       setBlocker(t, null);
   }

将一个线程进行挂起是通过park方法实现的,调用 park后,线程将一直阻塞直到超时或者中断等条件出现。unpark可以终止一个挂起的线程,使其恢复正常。整个并发框架中对线程的挂起操作被封装在 LockSupport类中,LockSupport类中有各种版本pack方法,但最终都调用了Unsafe.park()方法。

4.可重入锁

本文里面讲的是广义上的可重入锁,而不是单指JAVA下的ReentrantLock

可重入锁,也叫做递归锁,指的是同一线程外层函数获得锁之后 ,内层递归函数仍然有获取该锁的代码,但不受影响。在JAVA环境下 ReentrantLock 和synchronized 都是可重入锁。

synchronized

package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;/**
* <p>
*
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年02月13日 10:14
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年02月13日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class SynchronizedLock implements Runnable{    /**
    * When an object implementing interface <code>Runnable</code> is used
    * to create a thread, starting the thread causes the object's
    * <code>run</code> method to be called in that separately executing
    * thread.
    * <p>
    * The general contract of the method <code>run</code> is that it may
    * take any action whatsoever.
    *
    * @see Thread#run()
    */
   @Override
   public void run() {
       get();    
   }    private synchronized void get() {
       log.debug("synchronized current thread's id is {}",Thread.currentThread().getId());
       set();
   }    private synchronized void set() {
       log.debug("synchronized current thread's id is {}",Thread.currentThread().getId());
   }    public static void main(String[] args) {
       SynchronizedLock lock = new SynchronizedLock();        new Thread(lock).start();        new Thread(lock).start();        new Thread(lock).start();
   }
}

日志

10:22:31.186 [Thread-0] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 13
10:22:31.191 [Thread-0] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 13
10:22:31.191 [Thread-2] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 15
10:22:31.191 [Thread-2] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 15
10:22:31.191 [Thread-1] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 14
10:22:31.191 [Thread-1] DEBUG com.example.concurrence.lock.SynchronizedLock - synchronized current thread's id is 14

ReentrantLock

package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.locks.ReentrantLock;/**
* <p>
*
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年02月13日 10:19
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年02月13日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class ReentrantLockThread implements Runnable{   ReentrantLock lock = new ReentrantLock();    /**
    * When an object implementing interface <code>Runnable</code> is used
    * to create a thread, starting the thread causes the object's
    * <code>run</code> method to be called in that separately executing
    * thread.
    * <p>
    * The general contract of the method <code>run</code> is that it may
    * take any action whatsoever.
    *
    * @see Thread#run()
    */
   @Override
   public void run() {
       get();
   }    private void get() {
       lock.lock();
       log.debug("ReentrantLockThread current thread's id is {}",Thread.currentThread().getId());
       set();
       lock.unlock();
   }    private void set() {
       lock.lock();
       log.debug("ReentrantLockThread current thread's id is {}",Thread.currentThread().getId());
       lock.unlock();
   }    public static void main(String[] args) {
       ReentrantLockThread lock = new ReentrantLockThread();        new Thread(lock).start();        new Thread(lock).start();        new Thread(lock).start();
   }
}

日志

10:24:22.765 [Thread-0] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 13
10:24:22.770 [Thread-0] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 13
10:24:22.770 [Thread-1] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 14
10:24:22.770 [Thread-1] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 14
10:24:22.770 [Thread-2] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 15
10:24:22.770 [Thread-2] DEBUG com.example.concurrence.lock.ReentrantLockThread - ReentrantLockThread current thread's id is 15

两个例子最后的结果都是正确的,即 同一个线程id被连续输出两次。

可重入锁最大的作用是避免死锁,我们以自旋锁作为例子:

public class SpinLock {    private AtomicReference<Thread> owner =new AtomicReference<>();    public void lock(){
       Thread current = Thread.currentThread();        while(!owner.compareAndSet(null, current)){
       }
   }    public void unlock (){
       Thread current = Thread.currentThread();
       owner.compareAndSet(current, null);
   }
}

对于自旋锁来说, 1、若有同一线程两调用lock() ,会导致第二次调用lock位置进行自旋,产生了死锁 说明这个锁并不是可重入的。(在lock函数内,应验证线程是否为已经获得锁的线程) 2、若1问题已经解决,当unlock()第一次调用时,就已经将锁释放了。实际上不应释放锁。 (采用计数次进行统计)

修改之后

public class SpinLock1 {    private AtomicReference<Thread> owner =new AtomicReference<>();    private int count =0;    public void lock(){
       Thread current = Thread.currentThread();        if(current==owner.get()) {
           count++;            return ;
       }        while(!owner.compareAndSet(null, current)){       }
   }    public void unlock (){
       Thread current = Thread.currentThread();        if(current==owner.get()){            if(count!=0){
               count--;
           }else{
               owner.compareAndSet(current, null);
           }       }   }
}

该自旋锁即为可重入锁。

5.ReentrantLock(重入锁)以及公平性

ReentrantLock的实现不仅可以替代隐式的synchronized关键字,而且能够提供超过关键字本身的多种功能。 这里提到一个锁获取的公平性问题,如果在绝对时间上,先对锁进行获取的请求一定被先满足,那么这个锁是公平的,反之,是不公平的,也就是说等待时间最长的线程最有机会获取锁,也可以说锁的获取是有序的。ReentrantLock这个锁提供了一个构造函数,能够控制这个锁是否是公平的。 而锁的名字也是说明了这个锁具备了重复进入的可能,也就是说能够让当前线程多次的进行对锁的获取操作,这样的最大次数限制是Integer.MAX_VALUE,约21亿次左右。 事实上公平的锁机制往往没有非公平的效率高,因为公平的获取锁没有考虑到操作系统对线程的调度因素,这样造成JVM对于等待中的线程调度次序和操作系统对线程的调度之间的不匹配。对于锁的快速且重复的获取过程中,连续获取的概率是非常高的,而公平锁会压制这种情况,虽然公平性得以保障,但是响应比却下降了,但是并不是任何场景都是以TPS作为唯一指标的,因为公平锁能够减少“饥饿”发生的概率,等待越久的请求越是能够得到优先满足。

在ReentrantLock中,对于公平和非公平的定义是通过对同步器AbstractQueuedSynchronizer的扩展加以实现的,也就是在tryAcquire的实现上做了语义的控制。

非公平的获取语义

 final boolean nonfairTryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState();    if (c == 0) {        if (compareAndSetState(0, acquires)) {
           setExclusiveOwnerThread(current);            return true;
       }
   } else if (current == getExclusiveOwnerThread()) {        int nextc = c + acquires;                if (nextc < 0) // overflow
           throw new Error("Maximum lock count exceeded");
       setState(nextc);        return true;
   }    return false;
}

上述逻辑主要包括:

  • 如果当前状态为初始状态,那么尝试设置状态;
  • 如果状态设置成功后就返回;
  • 如果状态被设置,且获取锁的线程又是当前线程的时候,进行状态的自增;
  • 如果未设置成功状态且当前线程不是获取锁的线程,那么返回失败。

公平的获取语义

protected final boolean tryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState();    if (c == 0) {        if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
           setExclusiveOwnerThread(current);            return true;
       }
   } else if (current == getExclusiveOwnerThread()) {        int nextc = c + acquires;        if (nextc < 0)            throw new Error("Maximum lock count exceeded");
       setState(nextc);        return true;
   }    return false;
}

上述逻辑相比较非公平的获取,仅加入了当前线程(Node)之前是否有前置节点在等待的判断。hasQueuedPredecessors()方法命名有些歧义,其实应该是currentThreadHasQueuedPredecessors()更为妥帖一些,也就是说当前面没有人排在该节点(Node)前面时候队且能够设置成功状态,才能够获取锁。

释放语义

protected final boolean tryRelease(int releases) {    int c = getState() - releases;    if (Thread.currentThread() != getExclusiveOwnerThread())        throw new IllegalMonitorStateException();    boolean free = false;    if (c == 0) {
       free = true;
       setExclusiveOwnerThread(null);
   }
   setState(c);    return free;
}

上述逻辑主要主要计算了释放状态后的值,如果为0则完全释放,返回true,反之仅是设置状态,返回false。 下面将主要的笔墨放在公平性和非公平性上,首先看一下二者测试的对比:

package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;/**
* <p>
*
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年02月13日 10:39
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年02月13日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class ReentrantLockWithFailCondition {    private static Lock fairLock = new ReentrantLock(true);    private static Lock unfairLock = new ReentrantLock();    public void fair() {
       log.debug("fair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(fairLock));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    public void unfair() {
       log.debug("unfair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(unfairLock));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    private static class Job implements Runnable {        private Lock lock;        public Job(Lock lock) {            this.lock = lock;
       }        @Override
       public void run() {            for (int i = 0; i < 5; i++) {
               lock.lock();                try {
                   System.out.println("Lock by:"
                           + Thread.currentThread().getName());
               } finally {
                   lock.unlock();
               }
           }
       }
   }    public static void main(String[] args) {
       ReentrantLockWithFailCondition test = new ReentrantLockWithFailCondition();
       test.fair();
       log.debug("unfair --------------------------------------------------->>>");
       test.unfair();
   }
}

仔细观察返回的结果(其中每个数字代表一个线程),非公平的结果一个线程连续获取锁的情况非常多,而公平的结果连续获取的情况基本没有。那么在一个线程获取了锁的那一刻,究竟锁的公平性会导致锁有什么样的处理逻辑呢? 通过之前的同步器(AbstractQueuedSynchronizer)的介绍,在锁上是存在一个等待队列,sync队列,我们通过复写ReentrantLock的获取当前锁的sync队列,输出在ReentrantLock被获取时刻,当前的sync队列的状态。

修改测试

package com.example.concurrence.lock;import lombok.extern.slf4j.Slf4j;import java.util.Collection;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;/**
* <p>
*
* </p>
*
* @author xiachaoyang
* @version V1.2.0
* @date 2019年02月13日 10:39
* @modificationHistory=========================逻辑或功能性重大变更记录
* @modify By: {修改人} 2019年02月13日
* @modify reason: {方法名}:{原因}
* ...
*/@Slf4jpublic class ReentrantLockWithFailCondition{    private static Lock fairLock = new ReentrantLock(true);    private static Lock unfairLock = new ReentrantLock();    public void fair() {
       log.debug("fair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(fairLock));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    public void unfair() {
       log.debug("unfair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(unfairLock));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    private static class Job implements Runnable {        private Lock lock;        public Job(Lock lock) {            this.lock = lock;
       }        @Override
       public void run() {            for (int i = 0; i < 5; i++) {
               lock.lock();                try {                    //log.debug("Lock by:{}",Thread.currentThread().getName());
                   log.debug("Lock by:{} and {} waits.",Thread.currentThread().getName(),((ReentrantLock2) lock).getQueuedThreads());
               } finally {
                   lock.unlock();
               }
           }
       }
   }    private static Lock fairLock2 = new ReentrantLock2(true);    private static Lock unfairLock2 = new ReentrantLock2();    private static class ReentrantLock2 extends ReentrantLock {        // Constructor Override
       private static final long serialVersionUID = 1773716895097002072L;        /**
        * Creates an instance of {@code ReentrantLock} with the
        * given fairness policy.
        *
        * @param fair {@code true} if this lock should use a fair ordering policy
        */
       public ReentrantLock2(boolean fair) {            super(fair);
       }        /**
        * Creates an instance of {@code ReentrantLock}.
        * This is equivalent to using {@code ReentrantLock(false)}.
        */
       public ReentrantLock2() {
       }        @Override
       public Collection<Thread> getQueuedThreads() {            return super.getQueuedThreads();
       }
   }    public void fair2() {
       log.debug("fair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(fairLock2));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    public void unfair2() {
       log.debug("unfair version");        for (int i = 0; i < 5; i++) {
           Thread thread = new Thread(new Job(unfairLock2));
           thread.setName("" + i);
           thread.start();
       }        try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }    public static void main(String[] args) {
       ReentrantLockWithFailCondition test = new ReentrantLockWithFailCondition();       // test.fair();
       //log.debug("unfair --------------------------------------------------->>>");
       //test.unfair();       test.fair2();
       log.debug("unfair2 --------------------------------------------------->>>");
       test.unfair2();
   }
}

可以明显看出,在非公平获取的过程中,“插队”现象非常严重,后续获取锁的线程根本不顾及sync队列中等待的线程,而是能获取就获取。反观公平获取的过程,锁的获取就类似线性化的,每次都由sync队列中等待最长的线程(链表的第一个,sync队列是由尾部结点添加,当前输出的sync队列是逆序输出)获取锁。一个 hasQueuedPredecessors方法能够获得公平性的特性,

这点实际上是由AbstractQueuedSynchronizer来完成的,看一下acquire方法:

public final void acquire(int arg) {    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
       selfInterrupt();
}

可以看到,如果获取状态和在sync队列中排队是短路的判断,也就是说如果tryAcquire成功,那么是不会进入sync队列的,可以通过下图来深刻的认识公平性和AbstractQueuedSynchronizer的获取过程。 非公平的,或者说默认的获取方式如下图所示:

对于状态的获取,可以快速的通过tryAcquire的成功,也就是黄色的Fast路线,也可以由于tryAcquire的失败,构造节点,进入sync队列中排序后再次获取。因此可以理解为Fast就是一个快速通道,当例子中的线程释放锁之后,快速的通过Fast通道再次获取锁,就算当前sync队列中有排队等待的线程也会被忽略。这种模式,可以保证进入和退出锁的吞吐量,但是sync队列中过早排队的线程会一直处于阻塞状态,造成“饥饿”场景。 而公平性锁,就是在tryAcquire的调用中顾及当前sync队列中的等待节点(废弃了Fast通道),也就是任意请求都需要按照sync队列中既有的顺序进行,先到先得。这样很好的确保了公平性,但是可以从结果中看到,吞吐量就没有非公平的锁高了。

REFERENCES

  • [java锁的种类以及辨析(一):自旋锁] (http://ifeve.com/java_lock_see1/)
  • [Java锁的种类以及辨析(二):自旋锁的其他种类] (http://ifeve.com/java_lock_see2/)
  • [Java锁的种类以及辨析(三):阻塞锁] (http://ifeve.com/java_lock_see3/)
  • [Java锁的种类以及辨析(四):可重入锁] (http://ifeve.com/java_lock_see4/)
  • [ReentrantLock(重入锁)以及公平性] (http://ifeve.com/reentrantlock-and-fairness/)
  • [Java多线程系列—“JUC原子类”04之 AtomicReference原子类] (http://www.cnblogs.com/skywang12345/p/3514623.html)
  • [浅析TicketLock] (https://blog.csdn.net/yxc5463/article/details/78193991)
  • [Java多线程编程排队锁(Ticket Lock详解)] (http://www.leftso.com/blog/466.html)
  • [Java中Unsafe类详解] (https://www.cnblogs.com/mickole/articles/3757278.html)

原文发布于微信公众号 - 架构探险之道(zacsnz1314)

原文发表时间:2019-02-14

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券