每日一博 | 有助于提高"锁"性能的几种方法（附源码示例）

最近阅读《java高并发编程一书》大概总结几条，也是书中的内容

1.减小锁持有的时间

比如100个人去银行办理业务，要填一百张表，但是只有一支笔，那么很显然，每个人用笔的时间越短，效率也就月高：看代码：

/*
othercode1和othercode2很耗时间,里面没有涉及资源同步,只有mutexMethod方法要对资源同步,
所有优化代码让持有锁时间尽量短
*/

public synchronized void syncMethod(){
        othercode1();
        mutexMethod();
        othercode2();
}

public  void syncMethod(){
        othercode1();
        synchronized(this){
            mutexMethod();
        }
        othercode2();
}
//在jdk源码里面也很容易找到这种手段,比如处理正则表达式的Pattern类
public Matcher matcher(CharSequence input) {
        if (!compiled) {
            synchronized(this) {
                if (!compiled)
                    compile();
            }
        }
        Matcher m = new Matcher(this, input);
        return m;
}
//只有在表达式未编译的时候进行局部加锁,这种方法大大提高了matcher的执行效率和可靠性

注意:减少锁的持有时间有助于降低锁冲突的可能性,进而提升系统的并发能力

2.减小锁的力度

concurrentHashMap的实现,他的内部被分为了若干个晓得hashmap,称之为段(SEGMENT),默认是16段

减小锁粒度会引入一个新的问题,当需要获取全局锁的时候,其消耗的资源会较多,不如concurrenthashMap的size()方法.可以看到计算size的时候需要计算全部有效的段的锁

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
}

final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

事实上计算size的时候会先使用无锁的方式计算,如果失败会采用这个方法,但是在高并发的场合concurrenthashmap的size依然要差于同步的hashmap.因此在类似于size获取全局信息方法调用不频繁的情况下,这种减小粒度的的方法才是真正意义上的提高系统并发量

注意:所谓减小锁粒度,就是指缩小锁定对象的范围,从而减小锁冲突的可能性,进而提高系统性能

3.读写分离来替换独占锁

在读多写少的情况下,使用读写锁可以有效的提高系统性能 ReadWriteLock可以提高系统性能

package com.high.concurrency;

import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author: zhangzeli
 * @date 8:43 2018/4/10
 * <P></P>
 */
public class ReadWriteLockDemo {
    private static Lock lock = new ReentrantLock();
    private static ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private static Lock readLock =readWriteLock.readLock();
    private static Lock writeLock = readWriteLock.writeLock();
    private int value;

    public Object handleRead(Lock lock) throws InterruptedException{
        try {
            lock.lock();
            Thread.sleep(1000);
            return value;
        }finally {
            lock.unlock();
        }
    }

    public void handleWrite(Lock lock,int index) throws InterruptedException{
        try {
            lock.lock();
            Thread.sleep(1000);
            value =index;
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        final ReadWriteLockDemo demo = new ReadWriteLockDemo();
        Runnable readRunnale = new Runnable() {
            @Override
            public void run() {
                try {
                    demo.handleRead(lock);
                    //demo.handleRead(readLock);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        Runnable write = new Runnable() {
            @Override
            public void run() {
                try {
                    //demo.handleWrite(writeLock,new Random().nextInt());
                    demo.handleWrite(lock,new Random().nextInt());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        for(int i=0;i<18;i++){
            new Thread(readRunnale).start();
        }
        for(int i=18;i<20;i++){
            new Thread(write).start();
        }
    }
}

差异很明显.

3.锁分离

已LinkedBlockingQueue为例,take函数和put函数分别实现了冲队列取和往队列加数据，虽然两个方法都对队列进项了修改，但是LinkedBlockingQueue是基于链表的所以一个操作的是头，一个是队列尾端，从理论情况下将并不冲突

如果使用独占锁则take和put就不能完成真正的并发，所以jdk并没有才用这种方式取而代之的是两把不同的锁分离了put和take的操作，下面看源码

/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();//take函数需要持有takeLock

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();//put函数需要持有putLock

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly(); //不能有两个线程同时取数据
        try {
            while (count.get() == 0) { //如果当前没有可用数据,一直等待
                notEmpty.await();      //等待,put操作的通知
            }
            x = dequeue();         //取得第一个数据
            c = count.getAndDecrement();//数量减一,原子操作因为回合put同时访问count.注意变量c是count减一
            if (c > 1)
                notEmpty.signal();  //通知其他take操作
        } finally {
            takeLock.unlock(); //释放锁
        }
        if (c == capacity)
            signalNotFull(); //通知put,已有空余空间
        return x;
    }


public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();//不能有两个线程同时进行put
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) { //如果队列已满
                notFull.await();   //等待
            }
            enqueue(node);  //插入数据
            c = count.getAndIncrement(); //更新总数,变量c是count加1前的值
            if (c + 1 < capacity)
                notFull.signal();   //有足够的空间,通知其他线程
        } finally {
            putLock.unlock();  //释放锁
        }
        if (c == 0)
            signalNotEmpty();  //插入成功后,通知take操作
    }

4.锁粗化

虚拟机在遇到一连串地对同一锁不断进行请求和释放的操作时,便会把所有的锁操作整合成对锁的一次请求,从而减小对锁的请求同步次数,这个操作叫锁粗话,比如

for (int i=0;i<20;i++){
     synchronized (lock){
                
     }
}

//优化后
synchronized (lock){
    for (int i=0;i<20;i++){
     
                
     }
}

注意:性能优化就是根据运行时的真实情况对各个资源点进行权衡折中的过程,锁粗话的思想和减少锁持有时间是相反的,但是在不同的场合,他们的效果并不相同,所以大家要根据实际情况,进行权衡

5.java虚拟机对锁优化所做的努力

5.1锁偏向

偏向锁，简单的讲，就是在锁对象的对象头中有个ThreaddId字段，这个字段如果是空的，第一次获取锁的时候，就将自身的ThreadId写入到锁的ThreadId字段内，将锁头内的是否偏向锁的状态位置1.这样下次获取锁的时候，直接检查ThreadId是否和自身线程Id一致，如果一致，则认为当前线程已经获取了锁，因此不需再次获取锁，略过了轻量级锁和重量级锁的加锁阶段。提高了效率。

但是偏向锁也有一个问题，就是当锁有竞争关系的时候，需要解除偏向锁，使锁进入竞争的状态

参数-XX:+UseBiasedLocking

Java偏向锁(Biased Locking)是Java6引入的一项多线程优化。它通过消除资源无竞争情况下的同步原语，
进一步提高了程序的运行性能。偏向锁，顾名思义，它会偏向于第一个访问锁的线程，如果在接下来的运行过程中，
该锁没有被其他的线程访问，则持有偏向锁的线程将永远不需要触发同步。如果在运行过程中，遇到了其他线程抢占锁，
则持有偏向锁的线程会被挂起，JVM会尝试消除它身上的偏向锁，将锁恢复到标准的轻量级锁。(偏向锁只能在单线程下起作用)
因此 流程是这样的 偏向锁->轻量级锁->重量级锁

5.2轻量级锁

轻量级锁加锁：线程在执行同步块之前，JVM会先在当前线程的栈桢中创建用于存储锁记录的空间，并将对象头中的Mark Word复制到锁记录中，官方称为Displaced Mark Word。

然后线程尝试使用CAS将对象头中的Mark Word替换为指向锁记录的指针。如果成功，当前线程获得锁，如果失败，表示其他线程竞争锁，当前线程便尝试使用自旋来获取锁。

轻量级锁解锁：轻量级解锁时，会使用原子的CAS操作来将Displaced Mark Word替换回到对象头，如果成功，则表示没有竞争发生。

如果失败，表示当前锁存在竞争，锁就会膨胀成重量级锁。

注：轻量级锁会一直保持，唤醒总是发生在轻量级锁解锁的时候，因为加锁的时候已经成功CAS操作；而CAS失败的线程，会立即锁膨胀，并阻塞等待唤醒。（详见下图）

下图是两个线程同时争夺锁，导致锁膨胀的流程图。

锁不会降级

自旋其实就是虚拟机为了避免线程真实的在操作系统层挂起,虚拟机让当前线程做空轮询或许是几个cpu时间周期,如果还没办法获取锁则在挂起.

因为自旋会消耗CPU，为了避免无用的自旋（比如获得锁的线程被阻塞住了），一旦锁升级成重量级锁，
就不会再恢复到轻量级锁状态。
当锁处于这个状态下，其他线程试图获取锁时，都会被阻塞住，当持有锁的线程释放锁之后会唤醒这些线程，
被唤醒的线程就会进行新一轮的夺锁之争。

5.3锁消除

锁消除是Java虚拟机在JIT编译是，通过对运行上下文的扫描，去除不可能存在共享资源竞争的锁，通过锁消除，可以节省毫无意义的请求锁时间

public class TestLockEliminate {
    public static String getString(String s1, String s2) {
        StringBuffer sb = new StringBuffer();
        sb.append(s1);
        sb.append(s2);
        return sb.toString();
    }

    public static void main(String[] args) {
        long tsStart = System.currentTimeMillis();
        for (int i = 0; i < 1000000; i++) {
            getString("TestLockEliminate ", "Suffix");
        }
        System.out.println("一共耗费：" + (System.currentTimeMillis() - tsStart) + " ms");
    }
}

getString()方法中的StringBuffer数以函数内部的局部变量，进作用于方法内部，不可能逃逸出该方法，因此他就不可能被多个线程同时访问，也就没有资源的竞争，但是StringBuffer的append操作却需要执行同步操作:

@Override
public synchronized StringBuffer append(String str) {
        toStringCache = null;
        super.append(str);
        return this;
}

逃逸分析和锁消除分别可以使用参数-XX:+DoEscapeAnalysis和-XX:+EliminateLocks(锁消除必须在-server模式下)开启。使用如下参数运行上面的程序：

使用如下命令运行程序：-XX:+DoEscapeAnalysis -XX:+EliminateLocks

锁的优缺点对比