皮皮网

1.【Java原理系列】Java AtomicInteger原理用法源码详解
2.ForkjoinPool -1
3.编程「锁」事｜详解乐观锁 CAS 的例源技术原理
4.从HotSpot源码，深度解读 park 和 unpark
5.深入学习CAS底层原理

【Java原理系列】Java AtomicInteger原理用法源码详解

       Java的源码原子类AtomicInteger，是解析《Java原理用法示例及代码规范详解系列》的一部分，关注和收藏以获取最新内容。例源它用于在多线程环境中进行安全的源码整数操作，如get(),解析源码写成脚本 set(), incrementAndGet(), compareAndSet()等，提高并发性能，例源适用于计数器、源码标记位等场景。解析

       AtomicInteger的例源核心原理基于CAS操作，内部使用volatile修饰的源码int变量保证可见性和原子性。CAS操作确保在多线程环境中，解析对整数的例源修改是原子性的，避免了竞态条件和数据不一致。源码如果CAS操作失败，解析它会通过循环重试确保操作成功。

       在使用AtomicInteger时，如计数器递增和条件判断，应避免竞态条件。通过额外的同步手段如锁或Lock接口，可以确保整个操作序列是原子的。AtomicInteger提供的方法如getAndIncrement()，保证了这些操作的线程安全。

       场景上，哈尔滨系统源码AtomicInteger在计数器、并发任务处理和共享变量的线程安全操作中大显身手。例如，网站访问计数和任务完成数量统计，AtomicInteger确保了这些操作的原子性，输出的计数始终准确。

       总的来说，AtomicInteger是处理多线程整数操作的理想选择，为并发编程提供了一种高效且线程安全的解决方案。

ForkjoinPool -1

        ForkJoin是用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。

        下面是一个是一个简单的Join/Fork计算过程，将1—数字相加

        通常这样个模型，你们会想到什么？

        Release Framework ？ 常见的处理模型是什么？ task pool - worker pool的模型。 但是Forkjoinpool 采取了完全不同的模型。

        ForkJoinPool一种ExecutorService的实现，运行ForkJoinTask任务。ForkJoinPool区别于其它ExecutorService，主要是因为它采用了一种工作窃取(work-stealing)的机制。所有被ForkJoinPool管理的线程尝试窃取提交到池子里的任务来执行，执行中又可产生子任务提交到池子中。

        ForkJoinPool维护了一个WorkQueue的数组(数组长度是2的整数次方，自动增长)。每个workQueue都有任务队列(ForkJoinTask的数组)，并且用base、top指向任务队列队尾和队头。work-stealing机制就是工作线程挨个扫描任务队列，如果队列不为空则取队尾的任务并执行。示意图如下

        流程图：

        pool属性

        workQueues是pool的属性，它是WorkQueue类型的数组。externalPush和externalSubmit所创建的workQueue没有owner(即不是worker)，且会被放到workQueues的偶数位置；而createWorker创建的workQueue（即worker）有owner，且会被放到workQueues的奇数位置。

        WorkQueue的几个重要成员变量说明如下：

        这是WorkQueue的config，高位跟pool的config值保持一致，而低位则是workQueue在workQueues数组的位置。

        从workQueues属性的介绍中，我们知道，不是所有workQueue都有worker，没有worker的workQueue称为公共队列（shared queue），config的第位就是用来判断是否是公共队列的。在externalSubmit创建工作队列时，有：

        q.config = k | SHARED_QUEUE;

        其中q是新创建的workQueue，k就是q在workQueues数组中的位置，SHARED_QUEUE=1<<，注意这里config没有保留mode的信息。

        而在registerWorker中，则是这样给workQueue的config赋值的：

        w.config = i | mode;

        w是新创建的workQueue，i是其在workQueues数组中的位置，没有设置SHARED_QUEUE标记位

        scanState是workQueue的属性，是int类型的。scanState的低位可以用来定位当前worker处于workQueues数组的哪个位置。每个worker在被创建时会在其构造函数中调用pool的registerWorker，而registerWorker会给scanState赋一个初始值，这个值是奇数，因为worker是由createWorker创建，并会被放到WorkQueues的奇数位置，而createWorker创建worker时会调用registerWorker。

        简言之，worker的scanState初始值是奇数，非worker的scanstate初始值=INACTIVE=1<<，小于0（非worker的workQueue在externalSubmit中创建）。

        当每次调用signalWork（或tryRelease）唤醒worker时，worker的高位就会加1

        另外，scanState<0表示worker未激活，当worker调用runtask执行任务时，scanState会被置为偶数，即设置scanState的最右边一位为0。

        worker休眠时，是这样存储的

        worker的唤醒类似这样：

        在worker休眠的4行伪码中，让ctl的低位的值变为worker.scanState，这样下次就可以通过scanState唤醒该worker。唤醒该worker时，把该worker的preStack设置为ctl低位的值，这样下下次唤醒的worker就是scanState等于该preStack的worker。

        这里通过preStack保存下一个worker，这个worker比当前worker更早地在等待，所以形成一个后进先出的栈。

        runState是int类型的值，控制整个pool的运行状态和生命周期，有下面几个值（可以好几个值同时存在）：

        如果runState值为0，表示pool尚未初始化。

        RSLOCK表示锁定pool，当添加worker和pool终止时，就要使用RSLOCK锁定整个pool。如果由于runState被锁定，导致其他操作等待runState解锁（通常用wait进行等待），当runState设置了RSIGNAL，表示runState解锁，并通知（notifyAll）等待的操作。

        剩下4个值都跟runState生命周期有关，都可以顾名思义：

        当需要停止时，设置runState的STOP值，表示准备关闭，这样其他操作看到这个标记位，就不会继续操作，比如tryAddWorker看到STOP就不会再创建worker：

        而tryTerminate对这些生命周期状态的处理则是这样的：

        当前top和base的初始值为 INITIAL_QUEUE_CAPACITY >>>1= (1 << )>>>1 = /2。然后push一个task之后，top+=1，也就是说，top对应的位置是没有task的，最近push进来的task在top-1的位置。而base的位置则能对应到task，base对应最先放进队列的task，top-1对应最后放进队列的task。

        qlock值含义：1: locked, < 0: terminate; else 0

        即当qlock值位0时，可以正常操作，值=1时，表示锁定

        int SQMASK=0xe，则任何整数跟SQMASK位与后，得到的数就是偶数。

        证明：

        注意这里化为二进制是 ，尤其注意最右边第一位是0，任何数跟最右边第一位是0的数位与后，得到的数就是偶数，因为位与之后，第一位就是0，比如s=A&SQMASK，A可以是任意整数，然后把s按二进制进行多项式展开，则有s=2 n1+2 n2 ……+2^nn，这里n≥1，所以s可以被2整除，即s是偶数。

        所以一个数是奇数还是偶数，看其最右边第一位即可。

        我们知道workQueue有externalPush创建的和createWorker创建的worker，两种方式创建的workQueue，其放置到workQueues的位置是不同的，前者放到workQueue的偶数位置，而后者则放到奇数位置。不同workQueue找到自己在workQueues的位置的算法有点不同。

        下面看一下forkjoin框架获取workQueues中的偶数位置的workQueue的算法：

        这样就能获取workQueues的偶数位置的workQueue。m保证m & r & SQMASK这整个运算结果不会超出workQueues的下标，SQMASK保证取到的是偶数位置的workQueue。这里有一个有趣的现象，假设0到workQueues.length-1之间有n个偶数，m & r & SQMASK每次都能取到其中一个偶数，而且连续n次取到的偶数不会出现重复值，散列性非常好。而且是循环的，即1到n次取n个不同偶数，n+1到2n也是取n次不同偶数，此时n个偶数每个都被重新取一次。下面分析下r值有什么秘密，为何能保证这样的散列性

        ThreadLocalRandom内有一常量PROBE_INCREMENT = 0x9eb9，以及一个静态的probeGenerator =new AtomicInteger() ，然后每个线程的probe= probeGenerator.addAndGet(PROBE_INCREMENT)所以第一个线程的probe值是0x9eb9，第二个线程的值就是0x9eb9+0x9eb9，第三个线程的值就是0x9eb9+0x9eb9+0x9eb9以此类推，整个值是线性的，可以用y=kx表示，其中k=0x9eb9，x表示第几个线程。这样每个线程的probe可以保证不一样，而且具有很好的离散性。

        实际上，可以不用0x9eb9这个值，用任意一个奇数都是可以的，比如1。如果用1的话，probe+=1，这样每个线程的probe就都是不同的，而且具有很好的离散性。也就是说，假设有限制条件probe<n，超过n则产生溢出。则probe自加n次后才会开始出现重复值，n次前probe每次自加的值都不同。实际上用任意一个奇数，都可以保证probe自加n次后才会开始出现重复值，有兴趣可看本文最后附录部分。由于奇数的离散性，所以只要线程数小于m或者SQMASK两者中的最小值，则每个线程都能唯一地占据一个ws中的一个位置

        当一个操作是在非ForkjoinThread的线程中进行的，则称该操作为外部操作。比如我们前面执行pool.invoke，invoke内又执行externalPush。由于invoke是在非ForkjoinThread线程中进行的（这里是在main线程中进行），所以是一个外部操作，调用的是externalPush。之后task的执行是通过ForkJoinThread来执行的，所以task中的fork就是内部操作，调用的是push，把任务提交到工作队列。其实fork的实现是类似下面这样的：

        即fork会根据执行自身的线程是否是ForkJoinThread的实例来判断是处于外部还是内部。那为何要区分内外部？

        任何线程都可以使用ForkJoin框架，但是对于非ForkJoinThread的线程，它到底是怎样的，ForkJoin无法控制，也无法对其优化。因此区分出内外部，这样方便ForkJoin框架对任务的执行进行控制和优化

        forkJoinPool.invoke(task)是把任务放入工作队列，并等待任务执行。源码如下

        这里externalPush负责任务提交，externalPush源码如下：

编程「锁」事｜详解乐观锁 CAS 的技术原理

       本文深入探讨乐观锁的核心实现方式——CAS（Compare And Swap）技术原理。CAS是一种在多线程环境下实现同步功能的机制，相较于悲观锁的加锁操作，CAS允许在不使用锁的情况下实现多线程间的变量同步。Java的并发包中的原子类正是利用CAS实现乐观锁。

       CAS操作包含三个操作数：需要更新的内存值V、进行比较的预期数值A和要写入的值B。其逻辑是将内存值V与预期值A进行比较，当且仅当V值等于A时，通过原子方式用新值B更新V值（“比较+更新”整体是一个原子操作），否则不执行任何操作。一般情况下，更新操作会不断重试直至成功。

       以Java.util.concurrent.atomic并发包下的口令类源码AtomicInteger原子整型类为例，分析其CAS底层实现机制。方法`atomicData.incrementAndGet()`内部通过Unsafe类实现。Unsafe类是底层硬件CPU指令复制工具类，关键在于compareAndSet（）方法的返回结果。

       `unsafe.compareAndSwapInt(this, valueOffset, expect, update)`

       此方法中，参数`this`是Unsafe对象本身，用于获取value的内存偏移地址。`valueOffset`是value变量的内存偏移地址，`expect`是期望更新的值，`update`是要更新的最新值。如果原子变量中的value值等于`expect`，则使用`update`值更新该值并返回true，否则返回false。

       至于`valueOffset`的来源，这里提到value实际上是volatile关键字修饰的变量，以保证在多线程环境下的内存可见性。

       CAS的底层是Unsafe类。如何通过`Unsafe.getUnsafe()`方法获得Unsafe类的实例？这是因为AtomicInteger类在rt.jar包下，因此通过Bootstrap根类加载器加载。Unsafe类的具体实现可以在hotspot源码中找到，而unsafe.cpp中的C++代码不在本文详细分析范围内。对CAS实现感兴趣的读者可以自行查阅。

       CAS底层的nit 函数源码Unsafe类在多处理器上运行时，为cmpxchg指令添加lock前缀（lock cmpxchg），在单处理器上则无需此步骤（单处理器自身维护单处理器内的顺序一致性）。这一机制确保了CAS操作的原子性。

       最后，同学们会发现CAS的操作与原子性密切相关。CPU如何实现原子性操作是一个深入的话题，有机会可以继续探索。欢迎在评论区讨论，避免出现BUG！点赞转发不脱发！

从HotSpot源码，深度解读 park 和 unpark

       我最近建立了一个在线自习室（App：番茄ToDO）用于相互监督学习，感兴趣的小伙伴可以加入。自习室加入码：D5A7A

       Java并发包下的类大多基于AQS（AbstractQueuedSynchronizer）框架实现，而AQS线程安全的实现依赖于两个关键类：Unsafe和LockSupport。

       其中，Unsafe主要提供CAS操作（关于CAS，在文章《读懂AtomicInteger源码（多线程专题）》中讲解过），LockSupport主要提供park/unpark操作。实际上，park/unpark操作的最终调用还是基于Unsafe类，因此Unsafe类才是核心。

       Unsafe类的谷歌源码python实现是由native关键字说明的，这意味着这个方法是原生函数，是用C/C++语言实现的，并被编译成了DLL，由Java去调用。

       park函数的作用是将当前调用线程阻塞，而unpark函数则是唤醒指定线程。

       park是等待一个许可，unpark是为某线程提供一个许可。如果线程A调用park，除非另一个线程调用unpark(A)给A一个许可，否则线程A将阻塞在park操作上。每次调用一次park，需要有一个unpark来解锁。

       并且，unpark可以先于park调用，但不管unpark先调用多少次，都只提供一个许可，不可叠加。只需要一次park来消费掉unpark带来的许可，再次调用会阻塞。

       在Linux系统下，park和unpark是通过Posix线程库pthread中的mutex（互斥量）和condition（条件变量）来实现的。

       简单来说，mutex和condition保护了一个叫_counter的信号量。当park时，这个变量被设置为0，当unpark时，这个变量被设置为1。当_counter=0时线程阻塞，当_counter>0时直接设为0并返回。

       每个Java线程都有一个Parker实例，Parker类的部分源码如下：

       由源码可知，Parker类继承于PlatformParker，实际上是用Posix的mutex和condition来实现的。Parker类里的_counter字段，就是用来记录park和unpark是否需要阻塞的标识。

       具体的执行逻辑已经用注释标记在代码中，简要来说，就是检查_counter是不是大于0，如果是，则把_counter设置为0，返回。如果等于零，继续执行，阻塞等待。

       unpark直接设置_counter为1，再unlock mutex返回。如果_counter之前的值是0，则还要调用pthread_cond_signal唤醒在park中等待的线程。源码如下：

       （如果不会下载JVM源码可以后台回复“jdk”，获得下载压缩包）

深入学习CAS底层原理

       什么是CAS

       CAS是Compare-And-Swap的缩写，意思为比较并交换。以AtomicInteger为例，其提供了compareAndSet(intexpect,intupdate)方法，expect为期望值（被修改的值在主内存中的期望值），update为修改后的值。compareAndSet方法返回值类型为布尔类型，修改成功则返回true，修改失败返回false。

       举个compareAndSet方法的例子：

publicclassAtomticIntegerTest{ publicstaticvoidmain(String[]args){ AtomicIntegeratomicInteger=newAtomicInteger(0);booleanresult=atomicInteger.compareAndSet(0,1);System.out.println(result);System.out.println(atomicInteger.get());}}

       上面例子中，通过AtomicInteger(intinitialValue)构造方法指定了AtomicInteger类成员变量value的初始值为0：

publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable{ ......privatevolatileintvalue;/***CreatesanewAtomicIntegerwiththegiveninitialvalue.**@paraminitialValuetheinitialvalue*/publicAtomicInteger(intinitialValue){ value=initialValue;}......}

       接着执行compareAndSet方法，main线程从主内存中拷贝了value的副本到工作线程，值为0，并将这个值修改为1。如果此时主内存中value的值还是为0的话（言外之意就是没有被其他线程修改过），则将修改后的副本值刷回主内存更新value的值。所以上面的例子运行结果应该是true和1：

       将上面的例子修改为：

publicclassAtomticIntegerTest{ publicstaticvoidmain(String[]args){ AtomicIntegeratomicInteger=newAtomicInteger(0);booleanfirstResult=atomicInteger.compareAndSet(0,1);booleansecondResult=atomicInteger.compareAndSet(0,1);System.out.println(firstResult);System.out.println(secondResult);System.out.println(atomicInteger.get());}}

       上面例子中，main线程第二次调用compareAndSet方法的时候，value的值已经被修改为1了，不符合其expect的值，所以修改将失败。上面例子输出如下：

CAS底层原理

       查看compareAndSet方法源码：

/***Atomicallysetsthevalueto{ @codenewValue}*ifthecurrentvalue{ @code==expectedValue},*withmemoryeffectsasspecifiedby{ @linkVarHandle#compareAndSet}.**@paramexpectedValuetheexpectedvalue*@paramnewValuethenewvalue*@return{ @codetrue}ifsuccessful.Falsereturnindicatesthat*theactualvaluewasnotequaltotheexpectedvalue.*/publicfinalbooleancompareAndSet(intexpectedValue,intnewValue){ returnU.compareAndSetInt(this,VALUE,expectedValue,newValue);}

       该方法通过调用unsafe类的compareAndSwapInt方法实现相关功能。compareAndSwapInt方法包含四个参数：

       this，当前对象；

       valueOffset，value成员变量的内存偏移量（也就是内存地址）：

privatestaticfinallongvalueOffset;static{ try{ valueOffset=unsafe.objectFieldOffset(AtomicInteger.class.getDeclaredField("value"));}catch(Exceptionex){ thrownewError(ex);}}

       expect，期待值；

       update，更新值。

       所以这个方法的含义为：获取当前对象value成员变量在主内存中的值，和传入的期待值相比，如果相等则说明这个值没有被别的线程修改过，然后将其修改为更新值。

       那么unsafe又是什么？它的compareAndSwapInt方法是原子性的么？查看该方法的源码：

/***AtomicallyupdatesJavavariableto{ @codex}ifitiscurrently*holding{ @codeexpected}.**<p>Thisoperationhasmemorysemanticsofa{ @codevolatile}read*andwrite.CorrespondstoCatomic_compare_exchange_strong.**@return{ @codetrue}ifsuccessful*/@HotSpotIntrinsicCandidatepublicfinalnativebooleancompareAndSetInt(Objecto,longoffset,intexpected,intx);

       该方法并没有具体Java代码实现，方法通过native关键字修饰。由于Java方法无法直接访问底层系统，Unsafe类相当于一个后门，可以通过该类的方法直接操作特定内存的数据。Unsafe类存在于sun.msic包中，JVM会帮我们实现出相应的汇编指令。Unsafe类中的CAS方法是一条CPU并发原语，由若干条指令组成，用于完成某个功能的一个过程。原语的执行必须是连续的，在执行过程中不允许被中断，不会存在数据不一致的问题。

getAndIncrement方法剖析

       了解了CAS原理后，我们回头看下AtomicInteger的getAndIncrement方法源码：

/***Atomicallyincrementsthecurrentvalue,*withmemoryeffectsasspecifiedby{ @linkVarHandle#getAndAdd}.**<p>Equivalentto{ @codegetAndAdd(1)}.**@returnthepreviousvalue*/publicfinalintgetAndIncrement(){ returnU.getAndAddInt(this,VALUE,1);}

       该方法通过调用unsafe类的getAndAddInt方法实现相关功能。继续查看getAndAddInt方法的源码：

/***Atomicallyaddsthegivenvaluetothecurrentvalueofafield*orarrayelementwithinthegivenobject{ @codeo}*atthegiven{ @codeoffset}.**@paramoobject/arraytoupdatethefield/elementin*@paramoffsetfield/elementoffset*@paramdeltathevaluetoadd*@returnthepreviousvalue*@since1.8*/@HotSpotIntrinsicCandidatepublicfinalintgetAndAddInt(Objecto,longoffset,intdelta){ intv;do{ v=getIntVolatile(o,offset);}while(!weakCompareAndSetInt(o,offset,v,v+delta));returnv;}

       结合源码，我们便可以很直观地看出为什么AtomicInteger的getAndIncrement方法是线程安全的了：

       o是AtomicInteger对象本身；offset是AtomicInteger对象的成员变量value的内存地址；delta是需要变更的数量；v是通过unsafe的getIntVolatile方法获得AtomicInteger对象的成员变量value在主内存中的值。dowhile循环中的逻辑为：用当前对象的值和var5比较，如果相同，说明该值没有被别的线程修改过，更新为v+delta，并返回true（CAS）；否则继续获取值并比较，直到更新完成。

CAS的缺点

       CAS并不是完美的，其存在以下这些缺点：

       如果刚好while里的CAS操作一直不成功，那么对CPU的开销大；

       只能确保一个共享变量的原子操作；

       存在ABA问题。

       CAS实现的一个重要前提是需要取出某一时刻的数据并在当下时刻比较交换，这之间的时间差会导致数据的变化。比如：thread1线程从主内存中取出了变量a的值为A，thread2页从主内存中取出了变量a的值为A。由于线程调度的不确定性，这时候thread1可能被短暂挂起了，thread2进行了一些操作将值修改为了B，然后又进行了一些操作将值修改回了A，这时候当thread1重新获取CPU时间片重新执行CAS操作时，会发现变量a在主内存中的值仍然是A，所以CAS操作成功。

解决ABA问题

       那么如何解决CAS的ABA问题呢？由上面的阐述课件，光通过判断值是否相等并不能确保在一定时间差内值没有变更过，所以我们需要一个额外的指标来辅助判断，类似于时间戳，版本号等。

       JUC为我们提供了一个AtomicStampedReference类，通过查看它的构造方法就可以看出，除了指定初始值外，还需指定一个版本号（戳）：

/***Createsanew{ @codeAtomicStampedReference}withthegiven*initialvalues.**@paraminitialReftheinitialreference*@paraminitialStamptheinitialstamp*/publicAtomicStampedReference(VinitialRef,intinitialStamp){ pair=Pair.of(initialRef,initialStamp);}

       我们就用这个类来解决ABA问题，首先模拟一个ABA问题场景：

publicclassAtomticIntegerTest{ publicstaticvoidmain(String[]args){ AtomicReference<String>atomicReference=newAtomicReference<>("A");newThread(()->{ //模拟一次ABA操作atomicReference.compareAndSet("A","B");atomicReference.compareAndSet("B","A");System.out.println(Thread.currentThread().getName()+"线程完成了一次ABA操作");},"thread1").start();newThread(()->{ //让thread2先睡眠2秒钟，确保thread1的ABA操作完成try{ TimeUnit.SECONDS.sleep(2);}catch(InterruptedExceptione){ e.printStackTrace();}booleanresult=atomicReference.compareAndSet("A","B");if(result){ System.out.println(Thread.currentThread().getName()+"线程修改值成功，当前值为："+atomicReference.get());}},"thread2").start();}}

       运行程序，输出如下：

       使用AtomicStampedReference解决ABA问题：

publicclassAtomicIntegerextendsNumberimplementsjava.io.Serializable{ ......privatevolatileintvalue;/***CreatesanewAtomicIntegerwiththegiveninitialvalue.**@paraminitialValuetheinitialvalue*/publicAtomicInteger(intinitialValue){ value=initialValue;}......}0

       程序输出如下：