Java高并发(五)——Lock优化，提高性能

待我称王封你为后i 2022-04-11 11:20 217阅读 0赞

前边我们讲了，[多线程的世界][Link 1]，[多线程的基础操作][Link 2]，[多线程协作][Link 3]，[多线程管理——线程池][Link 4]。其中多线程为什么麻烦，就因为线程并行操作，对共享资源的争夺，会出现线程安全问题。而我们解决线程安全问题的方案是同步（锁资源，串行使用），串行就会出现性能问题。举个例子：大家在大道上并行前进的几列人（多线程并发），突然遇到河流，只有一个独木桥，大家只能一个个过（锁共享资源，串行使用）。显而易见，时间更多的消耗在过独木桥上，这也是性能瓶颈的所在。如何进行优化呢？这篇来看看，好，先看下这篇思维导图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70][]

可以思考下，我们可以从那几个方面优化？1，代码处理，即如何让线程更高效（安全的前提）的使用共享资源；2，JVM优化，如何通过经验，技巧更好的利用锁，提高执行效率；3，如何利用无锁（乐观锁代替悲观锁）替代锁，提高执行效率。而死锁会导致线程相互等待，造成非常严重后果。好，下边我们展开具体分析：

**一，代码——提高锁性能：**

**1，减小锁持有时间**：锁持有时间越长，相对的锁竞争程度也就也激烈。方法就是在我们编码时，只在必要的代码段进行同步，没有线程安全的代码，则不要进行锁。这样有助于降低锁冲突的可能性，进而提升系统的并发能力。

**2，减小锁粒度**：字面意思就是如何锁更小的资源。最典型的例子就是HashTable和ConcurrentHashMap，前者是锁住了整个对象，而后者只是锁住其中的要处理的Segment段（因为真正并发处理的这个Segment）。好比，人吃饭的，还可以看电视（锁住的是嘴，吃饭的时候不能喝茶，而不是整个人）。这里看下这个 [HashMap、HashTable和ConcurrentHashMap的文章][HashMap_HashTable_ConcurrentHashMap]（[http://www.cnblogs.com/-new/p/7496323.html][HashMap_HashTable_ConcurrentHashMap]）不了解的可以看下，更深刻理解JDK是如何高效的利用锁的。

**3，读写锁替换独占锁**：在读多写少的情况下，使用ReadWriteLock可以大大提高程序执行效率。相对容易理解不再赘述。

**4，锁分离**：读写锁分离了读操作和写操作，我们在进一步想，将更多的操作进行锁资源分离，就锁分离。典型的例子：LinkedBlockingQueue的实现，里边的take()和put()，虽然都操作整个队列进行修改数据，但是分别操作的队首和队尾，所以理论上是不冲突的。当然JDK是通过两个锁进行处理的，我们这里看下：

/** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock();
    
        /** Wait queue for waiting takes */
        private final Condition notEmpty = takeLock.newCondition();
    
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock();
    
        /** Wait queue for waiting puts */
        private final Condition notFull = putLock.newCondition();
    
        /**
         * put 操作
         * Inserts the specified element at the tail of this queue, waiting if
         * necessary for space to become available.
         *
         * @throws InterruptedException {@inheritDoc}
         * @throws NullPointerException {@inheritDoc}
         */
        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();
            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();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }
    
    
        /**
         * take 操作
         */     
        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();
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

** 5，锁粗化**：前边我们说了：减少锁持有时间、减小锁粒度。这里怎么又有锁粗化了。前者是从锁占有时间和范围的角度去考虑，这里我们从锁的请求、同步、释放的频率进行考虑，如果频率过高也会消耗系统的宝贵资源。典型场景：对于需要锁的资源在循环当中，我们可以直接将锁粗化到循环外层，而不是在内层（避免每次循环都申请锁、释放锁的操作）。

** 二，看下JVM对锁的优化**：

**1，偏向锁**：如果第一个线程获得了锁，则进行入偏向模式，如果接下来没有其他线程获取，则持有偏向锁的线程将不需要再进行同步。节省了申请所、释放锁，大大提高执行效率。如果锁竞争激烈的话，偏向锁将失效，还不如不用。可以通过-XX:+UseBiasedLocking进行控制开关。

** 2，轻量级锁**：将资源对象头部作为指针，指向持有锁的线程堆栈内部，来判断一个线程是否持有对象锁，如果线程获得（CAS）轻量级锁成功，则可以顺利进入同步块继续执行，否则轻量级锁失败，膨胀为重量级锁。  其提升程序同步性能的依据是：对于绝大部分的锁，在整个同步周期内都是不存在竞争的（前人经验数据）。

**3，自旋锁**：偏向锁和轻量级锁都失败了，JVM还不会立刻挂起此线程，JVM认为很快就可以得到锁，于是会做若干个空循环（自旋）来重新获取锁，如果获取锁成功，则进入同步块执行，否则，才会将其挂起。

**4，锁消除**：JVM在编译时，通过对运行上下文的扫描，去除不可能存在共享资源竞争的锁。我们在利用JDK一些类的时候，自己没有加锁，但是内部实现使用锁，而程序又不会出现并发安全问题，这时JVM就会帮我们进行锁消除，提高性能。例如：

//1,无锁代码
        private String spliceString(String s1,String s2,String s3){
            StringBuffer sb = new StringBuffer();
            sb.append(s1);
            sb.append(s2);
            sb.append(s3);
            return sb.toString()
        }
    
        //append的实现，这样JVM的锁消除就起作用了
        @Override
        public synchronized StringBuffer append(String str) {
            toStringCache = null;
            super.append(str);
            return this;
        }

**三，CAS——乐观锁**：悲观锁VS乐观锁，比较好理解，不再赘述。看下JDK通过CAS（Compare and swap）乐观锁实现的一些类。JDK的atomic包提供了一些直接使用CAS操作的线程安全的类。

**1，AtomicInteger**，通过CAS实现的一个线程安全的Integer，类似的还有AtomicBoolean、AtomicLong等。这里看下源码实现：

// setup to use Unsafe.compareAndSwapInt for updates
        private static final Unsafe unsafe = Unsafe.getUnsafe();
        private static final long valueOffset;
    
        static {
            try {
                valueOffset = unsafe.objectFieldOffset
                    (AtomicInteger.class.getDeclaredField("value"));
            } catch (Exception ex) { throw new Error(ex); }
        }
        
        //volatile修饰
        private volatile int value;
    
        /**
         * Atomically increments by one the current value.
         *
         * @return the updated value
         */
        public final int incrementAndGet() {
            return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
        }
    
        public final int getAndAddInt(Object var1, long var2, int var4) {
            int var5;
            do {
                var5 = this.getIntVolatile(var1, var2);
            } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
    
            return var5;
        }
    
        public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

**2，AtomicReference**：无锁的对象应用，和AtomicInteger非常相似的。底层源码实现也一直不在展开说。

**3，AtomicStampedReference**：带有时间戳的对象引用。上边几个Atomic类都是通过compare比较修改前和修改后的值是否一样，来保证数据正确的，但是如果就是出现了中间被人修改，但是又修改为原来的值（ABA），那么它们是不知道。而AtomicStampedReference通过添加时间戳解决了这个问题。

/**
         * 添加了新旧时间戳，数据库我们经常用version来做时间戳
         * Atomically sets the value of both the reference and stamp
         * to the given update values if the
         * current reference is {@code ==} to the expected reference
         * and the current stamp is equal to the expected stamp.
         *
         * @param expectedReference the expected value of the reference
         * @param newReference the new value for the reference
         * @param expectedStamp the expected value of the stamp
         * @param newStamp the new value for the stamp
         * @return {@code true} if successful
         */
        public boolean compareAndSet(V   expectedReference,
                                     V   newReference,
                                     int expectedStamp,
                                     int newStamp) {
            Pair<V> current = pair;
            return
                expectedReference == current.reference &&
                expectedStamp == current.stamp &&
                ((newReference == current.reference &&
                  newStamp == current.stamp) ||
                 casPair(current, Pair.of(newReference, newStamp)));
        }

**4，AtomicIntegerArray**：无锁数组，还有其他：AtomicLongArray、AtomicReferenceArray。看个简单例子吧：

//线程安全保证所有值都一样
    public class AtomicIntegerArrayDemo {
        static AtomicIntegerArray arr = new AtomicIntegerArray(10);
    
        public static class AddThread implements Runnable{
    
            @Override
            public void run() {
                for (int i = 0; i < 10000; i++) {
                    arr.getAndIncrement(i%arr.length());
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            Thread[] ts = new Thread[10];
            for(int k = 0;k<10;k++){
                ts[k] = new Thread(new AddThread());
            }
    
            for(int k=0;k<10;k++){
                ts[k].start();
            }
    
            for(int k=0;k<10;k++){
                ts[k].join();
            }
    
            System.out.println(arr);
        }
    }

**5，AtomicIntegerFieldUpdater**：让普通变量也享受原子操作，可以很容易帮我们以扩展的方式将原来不是线程安全的字段属性进行安全控制。还包括：AtomicLongFieldUpdater、AtomicReferenceFieldUpdater。看例子理解：

/**
     * 1,Updater只能修改它可见范围内的变量，因为是通过反射获取的，如果不可见就出错；
     * 2，为了确保变量被正确的读取，必须为volatile类型的；
     * 3，不支持static字段操作
     */
    public class AtomicIntegerFieldUpdaterDemo {
    
        public static class Candidate{
            int id;
            volatile int score;
        }
    
        public final static AtomicIntegerFieldUpdater<Candidate> scoreUpdater = AtomicIntegerFieldUpdater.newUpdater(Candidate.class,"score");
    
        public static AtomicInteger allSocre = new AtomicInteger(0);
    
        public static void main(String[] args) throws InterruptedException {
            final Candidate stu = new Candidate();
            Thread[] t =new Thread[10000];
            for (int i = 0; i < 10000; i++) {
                t[i] = new Thread(){
                    @Override
                    public void run() {
                        if(Math.random()>0.4){
                            scoreUpdater.incrementAndGet(stu);
                            allSocre.incrementAndGet();
                        }
                    }
                };
                t[i].start();
            }
            for (int i = 0; i < 10000; i++) {
                t[i].join();
            }
    
            System.out.println("score=" + stu.score);
            System.out.println("allscore=" + allSocre);
        }
    }

**四，避免死锁**：举个例子：喝水需要水壶和杯子，当两个线程出现一个持有水壶的锁，一个持有杯子的锁，而彼此都在等待彼此释放锁时，就会出现了死锁。出现死锁，线程就会一直永远等待，对程序而言是非常严重的bug。好看个例子：

public class DeadLock extends Thread {
    
        protected Object tool;
        static Object fork1 = new Object();
        static Object fork2 = new Object();
    
        public DeadLock(Object obj){
            this.tool = obj;
    
            if(tool == fork1){
                this.setName("哲学家A");
            }
            if(tool == fork2){
                this.setName("哲学家B");
            }
        }
    
    
        @Override
        public void run() {
            if(tool == fork1){
                synchronized (fork1){
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
    
                    synchronized (fork2){
                        System.out.println("哲学家A开始吃饭了");
                    }
                }
            }
    
            if(tool == fork2){
                synchronized (fork2){
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
    
                    synchronized (fork1){
                        System.out.println("哲学家B开始吃饭了");
                    }
                }
            }
        }
        public static void main(String[] args) throws InterruptedException {
            DeadLock zhexuejiaA = new DeadLock(fork1);
            DeadLock zhexuejiaB = new DeadLock(fork2);
            zhexuejiaA.start();
            zhexuejiaB.start();
    
            Thread.sleep(1000);
        }
    }

好了，多线程并发处理是提高了程序执行效率，但是带了资源安全问题。锁解决了资源安全问题，但是又有了性能问题，然后优化……代码级、jvm级、无锁……又没锁了，但是要求编码水平更高了，利用无锁实现类似有锁的功能，还是需要非常高的内功的，这样解决方案转转转……总是用合适的方法解决适合的问题。好了，多思考。继续中……

[Link 1]: https://blog.csdn.net/liujiahan629629/article/details/84039097
[Link 2]: https://blog.csdn.net/liujiahan629629/article/details/84205788
[Link 3]: https://blog.csdn.net/liujiahan629629/article/details/84312204
[Link 4]: https://blog.csdn.net/liujiahan629629/article/details/84454908
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70]: /images/20220411/a80866985a704f18950b762f36c28c00.png
[HashMap_HashTable_ConcurrentHashMap]: http://www.cnblogs.com/-new/p/7496323.html