垃圾回收算法与实现系列-JVM无锁实现

本是古典何须时尚 2022-10-12 04:12 108阅读 0赞

> **导语**  
>   为了确保多线程场景下数据安全，使用锁机制一直是一种优秀的解决方案，但是再高并发场景下，对锁的竞争可能成为性能瓶颈。为此，有出现了一种新的解决方案，被称为是非阻塞同步的方案。这种实现方式不需要使用锁，但依然可以保证高并发场景下数据的一致性，这一次就主要来说说这种不通过锁机制来实现数据安全的方案。

### 文章目录 ###

*   *  无锁机制
     *   *  理解CAS
         *  原子操作
         *  LongAdder

## 无锁机制 ##

### 理解CAS ###

基于锁的同步方式，也是一种阻塞的线程同步方式，无论是使用信号量、重入锁还是内部锁，受到核心资源的限制，不同的线程在锁竞争时总是不能避免等待，从而阻塞当前线程。为了避免出现这种问题，非阻塞同步的方式就被提出来。一种最简单的非阻塞同步的实现就是ThreadLocal，每一个线程拥有各自独立的变量副本，因此在并行计算的时候不需要相互等待。

下面介绍的CAS( Compare And Swap) 算法是一个比较常用的无锁并发控制方法，与锁机制实现相比较，无锁算法的设计与实现都复杂的很多，但由于其非阻塞性，它对死锁问题是先天性免疫，并且线程的相互影响也远远比基于锁的方式小。更重要的是，使用无锁的方式完全没有锁竞争带来的系统开销，也没有线程间频繁的上下文交换调度带来的开销，因此它要比基于锁机制的实现方式都更有优越性。

CAS算法的过程如下

包含三个参数，形式为CAS(V,E,N)。其中V表示要更新的变量，E表示预期值，N表示新值。仅当V的值等于E的值得时候，才会将V的值设置为N，如果V的值和E的值不同，说明已经有其他线程做了更新，则当前线程什么也不做。最后CAS返回当前V的真实值。CAS操作是抱着乐观的态度进行，它总是认为自己可以完成操作。当多个线程同时使用CAS操作一个变量的时候，只有一个线程可以操作成功，其余的线程都会失败。失败的线程不会被挂起，而是被告知失败，并且允许再次尝试，当然也允许失败的线程放弃选择。基于这样的一个操作原理，CAS操作即使没有锁，也可以发现其他线程对当前线程的干扰，并进行恰当的处理。

在硬件层面，大部分的处理器都支持了原子化出队CAS算法。在JDK1.5以后，虚拟机便可以使用这个指令来实现并发操作和并发的数据结构，并且这种操作在虚拟机中可以说无处不在。

### 原子操作 ###

为了能让CAS算被Java应用程序充分使用，在JDK的java.util.concurrent.atomic包下，提供了一组使用无锁算法实现的原子操作类，主要有AtomicInteger、AtomicIntegerArray、AtomicLong、AtomicLongArray等。分别分装了对整数、整数数组、长整型、长整型数组等多线程安全性操作。如下代码

public final int getAndSet(int newValue) { 
        return unsafe.getAndSetInt(this, valueOffset, newValue);
    }
    
    
     public final int getAndSetInt(Object var1, long var2, int var4) { 
            int var5;
            do { 
                var5 = this.getIntVolatile(var1, var2);
            } while(!this.compareAndSwapInt(var1, var2, var5, var4));
    
            return var5;
        }

在CAS算法中，首先是一个无穷循环，这个无穷循环用于多线程间的冲突处理，也就是在当前线程受其他线程影响而更新失败，就会不停地进行尝试，直到成功。

get()方法用于取得当前值，并使用compareAndSet()方法进行更新，如果未受其他线程影响，则预期值就等于current。可以将值更新为newValue，若更新成功，则退出循环。

如果受其他线程影响，则compareAndSet()时，预期值不等于current，更新失败，进行下一次循环，尝试继续更新，直到成功。

在整个更新过程中，无须加锁、无须等待。从这段代码中也可以看到，无锁的操作实际上将多线程并发的冲突交由应用层自行解决，这不仅仅提升了系统性能，还增加了系统的灵活性，但算法及编码的复杂度也明显地增加了。

public class Atomic{ 
    	private static final int MAX_THREADS = 3; //线程数
    	private static final int TASK_COUNT = 3;  // 任务数
    	private static final int TARGET_COUNT = 10000000;// 目标总数
        
        // 无锁的原子操作
        private AtomicLong acount = new AtomicLong（0L）
    	private long count = 0;
    	static CountDownLatch cdlsync = new CountDownLatch(TASK_COUNT);
    	static CountDownLatch cdlatomic = new CountDownLatch(TASK_COUNT);
    
    	protected synchronized long inc(){ 
    		return ++count;
    	}
    
    	protected synchronized long getCount(){ 
    		return count;
    	}
    
    	public void clearCount(){ 
    		count = 0;
    	}
    
    	public class SyncThread implements Runnable{ 
    		protected String name;
    		protected long starttime;
    		AtomicLess out;
    		public SyncThread(AtomicLess o,long starttime){ 
    			out = o;
    			this.starttime = starttime;
    		}
    
    		public void run(){ 
    			long v = out.getCount();
    			while(v<TARGET_COUNT){ 
    				v = out.inc();
    			}
    			long endtime = System.currentTimeMillis();
    			System.out.println("SyncThread spend :"+(endtime-starttime)+"ms"+v)
    			cdlsync.countDown();
    			
    		}
    	}
    	
    	public void testSync() throws InterruptedException{ 
    		ExecutorSerivce exe = Executors.newFixedThreadPool(MAX_THRE);
    		long starttime = System.currentTimeMillis();
    		SyncThread sync = new SyncThread(this,starttime);
    		for(int i = 0;i<TASK_COUNT;i++){ 
    			exe.submit(sync);
    		}
    		cdlsync.await();
    		exe.shutdown();
    	}
    	public class AtomicThread implements Runnable{ 
    		protected String name;
    		protected long starttime;
    		public AtomicThread(long starttime){ 
    			this.starttime = starttime;
    		}
    
    		public void run(){ 
    			long v = acount.get();
    			while(v<TARGET_COUNT){ 
    				v = acount.incremenetAndGet();
    			}
    			long endtime = System.currrentTimeMillis();
    			System.out.println("AtomicThread spend" + endtime-starttime)
    			cdlatomic.countDown();
    		}
    	}
    
    	public void testAtomic() throws InterrputedException{ 
    		ExecuteorService exe = Executors.newFixedThreadPool(MAX_THREADS);
    		long starttime = System.currentTimeMillis();
    		AtomicThread atomic = new AtomicThread(starttime);
    		for(int i =0;i<TASK_COUNT;i++){ 
    			exe.submit(atomic);
    		}
    
    		cdlatomic.await();
    		exe.shutdown();
    			
    	}
    
    	public static void mian(String args[]) throws InterruptedException{ 
    
    		AtomicLess a = new AtomicLess();
    		a.testSync();
    		a.testAtomic();
    	}
        
    }

上面代码测试了使用有锁的方式和无锁的方式对同一个数字进行的操作。根据程序中的设计，两种方式都使用了3个线程并行技术。有锁的操作将计数值保存在count变量中，无锁的原子将计数值保存在acount中。分别进行操作。根据结果分析可以知道，使用原子操作对计数器进行累加操作时，性能远远高于通过锁的方式进行的操作。这也是CAS算法在JVM内部被广泛使用的原因之一。

### LongAdder ###

上文中已经提到，无锁的原子类操作使用系统的CAS算法指令，性能上远远超过了加锁算法，那么是否可以存在一种更为高级的算法呢，在JDK1.8 中引入了LongAdder类，这个类也在并发原子包下，因此，可以推测它也使用了CAS算法指令。

前面提到的AtomicInteger等原子类的实现机制，都是在一个死循环内，不断尝试修改目标值，直到修改成功，如果竞争不激烈，那么修改成功的概率就高，那么如果竞争激烈的话，修改失败的概率就很高了。在大量的修改失败之后，就会进行多次的尝试，在性能上就有一定的损失。

结合之前提到的，减小锁粒度与ConcurrentHashMap 的实现，可以得到一点对于传统原子类的改进思路，虽然在CAS中没有锁，但是像减小锁的粒度这种分离热点的思想依然是适用的，一种可行的方案就是仿造ConcurrentHashMap，将热点数据分离，例如，将AtomicInteger内部核心数据value分离成一个数组，每个线程访问的时候，通过哈希算法映射到其中一个数字进行计数，最终的计数结果，则为这个数组的求和累加。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70_pic_center]  
  如图所示，热点数据value被分成了多个cell，每个cell独自维护内部的值，当前对象的实际值由所有的cell累计合成，这样热点就进行了有效的分离，从而提高了并行度，而LongAdder就是使用了这种思想。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/b3443c6af5a340139187fd2b2203152d.png