基于java中cas實(shí)現(xiàn)的探索

更新時(shí)間：2021年12月24日 08:49:46 作者：磨唧

這篇文章主要介紹了java中cas實(shí)現(xiàn)的探索，基于很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

1.背景簡介

當(dāng)我們在并發(fā)場景下，增加某個(gè)integer值的時(shí)，就涉及到多線程安全的問題，解決思路兩個(gè)

將值增加的方法使用同步代碼塊同步
使用AtomicInteger，來逐步增加其值

這兩種實(shí)現(xiàn)方式代碼如下

import java.util.concurrent.atomic.AtomicInteger;
public class CASTest {
    private static AtomicInteger countAI = new AtomicInteger(0);
    private static int count = 0;
    private static final int THREAD_COUNT = 8;
    public static void main(String[] args) throws InterruptedException {
        long start = System.currentTimeMillis();
        Thread[] threads = new Thread[THREAD_COUNT];
        for(int i = 0; i < THREAD_COUNT; i++) {
            threads[i] = new Thread(){
                @Override
                public void run() {
                    for(int i = 0; i < 10000000; i++) {
//                        測試1：使用同步代碼塊方法，耗時(shí)：2927ms
                        synAdd();
//                        測試2：使用atomicInterger方式, 耗時(shí)：1860ms
//                        atomicAdd();
                    }
                }
            };
        }
        for(int i = 0; i < THREAD_COUNT; i++) {
            threads[i].start();
        }
        for(int i = 0; i < THREAD_COUNT; i++) {
            threads[i].join();
        }
        System.out.println("finish...耗時(shí)：" + (System.currentTimeMillis()-start) + "ms");
    }
    private static synchronized void synAdd() {
        count++;
    }
    private static void atomicAdd() {
        countAI.getAndAdd(1);
    }
}

從測試結(jié)果可以看出，使用atomicAdd方法耗時(shí): 1860ms, 使用synAdd方法耗時(shí): 2927ms

為何使用AtomicInteger效率更高？以及AtomicInteger是如何實(shí)現(xiàn)的？本文將對cas進(jìn)行進(jìn)一步探索

2. java源碼追蹤

根據(jù)斷點(diǎn)追蹤countAI.getAndAdd(1);, 對棧如下

getAndAddInt:1034, Unsafe (sun.misc)
getAndAdd:177, AtomicInteger (java.util.concurrent.atomic)
atomicAdd:45, CASTest (com.youai.cas)
access$000:5, CASTest (com.youai.cas)
run:21, CASTest$1 (com.youai.cas)

進(jìn)入到了關(guān)鍵核心方法 sun.misc.Unsafe#getAndAddInt

    public final int getAndAddInt(Object o, long offset, int delta) {
        int v;
        do {
            v = getIntVolatile(o, offset);
        } while (!compareAndSwapInt(o, offset, v, v + delta));
        return v;
    }

在這個(gè)方法中，循環(huán)調(diào)用了sun.misc.Unsafe#compareAndSwapInt這個(gè)方法，這個(gè)方法的效果就是，判斷對象o中，地址偏移量是offset這個(gè)地址內(nèi)存中的int值是否和期望值v相等，如果相等，則用v + delta替換，并返回替換成功；否則不替換，并返回替換失敗。需要循環(huán)的原因是因?yàn)間etIntVolatile(o, offset);和compareAndSwapInt(o, offset, v, v + delta)這兩步并不是原子原作，在執(zhí)行前面一句后，目標(biāo)地址中的值可能被其他線程給修改，所以如果失敗需要重新獲取目標(biāo)地址中的最新值。

可以看到，在整個(gè)代碼過程中，并沒有強(qiáng)制加鎖，減少線程切換阻塞等無效時(shí)間的消耗，而是采用了失敗重試的機(jī)制，這也是樂觀鎖的一種實(shí)現(xiàn)。因?yàn)樗男矢摺?/p>

cas能夠?qū)崿F(xiàn)，需要compareAndSwapInt這個(gè)操作等價(jià)于一個(gè)原子操作，那compareAndSwapInt是如何實(shí)現(xiàn)的呢？下次解答。

3. hotspot jvm源碼追蹤

/**
     * Atomically update Java variable to <tt>x</tt> if it is currently
     * holding <tt>expected</tt>.
     * @return <tt>true</tt> if successful
     */
    public final native boolean compareAndSwapInt(Object o, long offset,
                                                  int expected,
                                                  int x);

可以看到compareAndSwapInt這個(gè)方法被native修飾，具體實(shí)現(xiàn)在需要參考c/c++代碼：

從openjdk源碼追蹤到compareAndSwapInt的實(shí)現(xiàn)在hotspot/src/share/vm/prims/unsafe.cpp這文件中, 具體對應(yīng)方法如下：

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
  UnsafeWrapper("Unsafe_CompareAndSwapInt");
  oop p = JNIHandles::resolve(obj);
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e; //此處調(diào)用了Atomic::cmpxchg方法
UNSAFE_END

Unsafe_CompareAndSwapInt方法進(jìn)一步調(diào)用了Atomic::cmpxchg方法，由于Atomic::cmpxchg方法和平臺有關(guān)，我們此時(shí)關(guān)注linux下的實(shí)現(xiàn)，hotspot/src/os_cpu/linux_x86/vm/atomic_linux_x86.inline.hpp，具體方法如下：

inline jint     Atomic::cmpxchg    (jint     exchange_value, volatile jint*     dest, jint     compare_value) {
  int mp = os::is_MP();
  __asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
                    : "=a" (exchange_value)
                    : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
                    : "cc", "memory");
  return exchange_value;
}

此方法是一個(gè)c++內(nèi)聯(lián)匯編的方法，我們著重關(guān)注cmpxchgl這個(gè)匯編指令：

在這里插入圖片描述

This instruction can be used with a LOCK prefix to allow the instruction to be executed atomically. To simplify the interface to the processor's bus, the destination operand receives a write cycle without regard to the result of the comparison. The destination operand is written back if the comparison fails; otherwise, the source operand is written into the destination. (The processor never produces a locked read without also producing a locked write.)

intel匯編指令的官方文檔來看， cmpxchgl的作用是，比較ax寄存器中的值和期望值，如果相等，則將target值設(shè)置到目標(biāo)對象上，否則不設(shè)置。特別得，在cmpxchgl指令前加上lock可以使得cmpxchgl操作成為一個(gè)原子操作。這也論證了sun.misc.Unsafe#compareAndSwapInt確是等價(jià)于一個(gè)原子操作

4. 手寫一個(gè)cas實(shí)現(xiàn)

1. 通過匯編手寫一個(gè)cas方法

看了intel的文檔，cas原理并不復(fù)雜，可以通過匯編手寫一個(gè)cas方法xchange：

	.file	"cmpandset.c"
	.text
	.globl	xchange
	.type	xchange, @function
xchange:
.LFB0:
	.cfi_startproc
	.cfi_def_cfa_offset 16
	.cfi_offset 6, -16
	.cfi_def_cfa_register 6
	mov %esi, %eax
	lock cmpxchgl %edx, (%rdi)
	sete %al
	movzbl %al, %eax
.L3:
	.cfi_def_cfa 7, 8
	ret
	.cfi_endproc
.LFE0:
	.size	xchange, .-xchange
	.ident	"GCC: (Ubuntu 7.5.0-3ubuntu1~18.04) 7.5.0"
	.section	.note.GNU-stack,"",@progbits

2. 多線程條件下測試自行實(shí)現(xiàn)的cas方法

測試代碼：

#include <stdio.h>
#include <pthread.h>
#include <sys/time.h>
#define THREAD_CNT 8
extern int xchange(int *ptr, int expect, int dest);
int a = 0;
void cmp_add(int* cnt, int adder);
long current_ms() {
    struct timeval cur_time;
    gettimeofday(&cur_time, NULL);
    return cur_time.tv_sec * 1000 + cur_time.tv_usec / 1000;
}
void * sum(void *arg) {
    for(int i = 0; i < 10000000; i++) {
        cmp_add(&a, 1);
    }
}
int main(int argc, char const *argv[])
{
    long start = current_ms();
    int result = xchange(&a, 13, 13);
    printf("result=%d\n", result);
    pthread_t tids[THREAD_CNT];
    for(int i = 0; i < THREAD_CNT; i++) {
        pthread_create(&tids[i], NULL, sum, NULL);
    }
    // 等待
    for(int i = 0; i < THREAD_CNT; i++) {
        pthread_join(tids[i], NULL);
    }
    printf("result=%d, 耗時(shí):%ldms\n", a, (current_ms() - start));
    
    return 0;
}
void cmp_add(int* cnt, int adder) {
    int tmp = 0;
    do {
        tmp = *cnt;
    } while(xchange(cnt, tmp, tmp+adder) == 0);
}

輸出結(jié)果為：

result=80000000, 耗時(shí):8596ms

可見自行實(shí)現(xiàn)的cas方法在多線程場景下，同樣是線程安全的。

3. cas與互斥鎖方式的對比

測試代碼：

#include <stdio.h>
#include <pthread.h>
#include <sys/time.h>
#include <semaphore.h>
#define THREAD_CNT 8
int a = 0;
sem_t add_mutex;
long current_ms() {
    struct timeval cur_time;
    gettimeofday(&cur_time, NULL);
    return cur_time.tv_sec * 1000 + cur_time.tv_usec / 1000;
}
void * sum(void *arg) {
    for(int i = 0; i < 10000000; i++) {
        sem_wait(&add_mutex);
        a++;
        sem_post(&add_mutex);
    }
}
int main(int argc, char const *argv[])
{
    long start = current_ms();
    
    sem_init(&add_mutex, 0, 1);
    pthread_t tids[THREAD_CNT];
    for(int i = 0; i < THREAD_CNT; i++) {
        pthread_create(&tids[i], NULL, sum, NULL);
    }
    
    // 等待
    for(int i = 0; i < THREAD_CNT; i++) {
        pthread_join(tids[i], NULL);
    }
    printf("result=%d, 耗時(shí):%ldms\n", a, (current_ms() - start));
    sem_destroy(&add_mutex);
    
    return 0;
}

輸出結(jié)果：