并發(fā)編程的目的是讓程序運(yùn)行更快，但是使用并發(fā)并不定會(huì)使得程序運(yùn)行更快，只有當(dāng)程序的并發(fā)數(shù)量達(dá)到一定的量級(jí)的時(shí)候才能體現(xiàn)并發(fā)編程的優(yōu)勢(shì)。所以談并發(fā)編程在高并發(fā)量的時(shí)候才有意義。雖然目前還沒(méi)有開(kāi)發(fā)過(guò)高并發(fā)量的程序，但是學(xué)習(xí)并發(fā)是為了更好理解一些分布式架構(gòu)。那么當(dāng)程序的并發(fā)量不高，比如是單線程的程序，單線程的執(zhí)行效率反而比多線程更高。這又是為什么呢？熟悉操作系統(tǒng)的應(yīng)該知道，CPU是通過(guò)給每個(gè)線程分配時(shí)間片的方式實(shí)現(xiàn)多線程的。這樣，當(dāng)CPU從一個(gè)任務(wù)切換到另一個(gè)任務(wù)的時(shí)候，會(huì)保存上一個(gè)任務(wù)的狀態(tài)，當(dāng)執(zhí)行完這個(gè)任務(wù)的時(shí)候CPU就會(huì)繼續(xù)上一個(gè)任務(wù)的狀態(tài)繼續(xù)執(zhí)行。這個(gè)過(guò)程稱(chēng)為上下文切換。

在Java多線程中，volatile關(guān)鍵字個(gè)synchronized關(guān)鍵字扮演了重要的角色，它們都可以實(shí)現(xiàn)線程的同步，但是在底層是如何實(shí)現(xiàn)的呢？

volatile

volatile只能保證變量對(duì)各個(gè)線程的可見(jiàn)性，但不能保證原子性。關(guān)于 Java語(yǔ)言 volatile 的使用方法就不多說(shuō)了，我的建議是 除了 配合package java.util.concurrent.atomic 中的類(lèi)庫(kù)，其他情況一概別用。更多的解釋 參見(jiàn) 這篇文章。

引子

參見(jiàn)如下代碼

 package org.go;

public class Go {

  volatile int i = 0;

  private void inc() {
    i++;
  }

  public static void main(String[] args) {
    Go go = new Go();
    for (int i = 0; i < 10; i++) {
      new Thread(() -> {
        for (int j = 0; j < 1000; j++)
          go.inc();
      }).start();
    }
    while(Thread.activeCount()>1){
      Thread.yield();
    }
    System.out.println(go.i);
  }
}
 

每次執(zhí)行上述代碼結(jié)果都不同，輸出的數(shù)字總是小于10000.這是因?yàn)樵谶M(jìn)行inc()的時(shí)候，i++并不是原子操作。或許有些人會(huì)提議說(shuō)用 synchronized 來(lái)同步inc() , 或者 用 package java.util.concurrent.locks 下的鎖去控制線程同步。但它們都不如下面的解決方案：

 package org.go;

import java.util.concurrent.atomic.AtomicInteger;

public class Go {

  AtomicInteger i = new AtomicInteger(0);

  private void inc() {
    i.getAndIncrement();
  }

  public static void main(String[] args) {
    Go go = new Go();
    for (int i = 0; i < 10; i++) {
      new Thread(() -> {
        for (int j = 0; j < 1000; j++)
          go.inc();
      }).start();
    }
    while(Thread.activeCount()>1){
      Thread.yield();
    }
    System.out.println(go.i);
  }
}

這時(shí)，如果你不了解 atomic 的實(shí)現(xiàn)，你一定會(huì)不屑的懷疑 說(shuō)不定 AtomicInteger 底層就是使用鎖來(lái)實(shí)現(xiàn)的所以也未必高效。那么究竟是什么，我們來(lái)看看。

原子類(lèi)的內(nèi)部實(shí)現(xiàn)

無(wú)論是AtomicInteger 或者是 ConcurrentLinkedQueue的節(jié)點(diǎn)類(lèi)ConcurrentLinkedQueue.Node,他們都有個(gè)靜態(tài)變量
 private static final sun.misc.Unsafe UNSAFE;,這個(gè)類(lèi)是實(shí)現(xiàn)原子語(yǔ)義的C++對(duì)象sun::misc::Unsafe的Java封裝。想看看底層實(shí)現(xiàn)，正好我手邊有g(shù)cc4.8的源代碼，對(duì)照本地路徑，很方便找到Github的路徑，看這里。

以接口 getAndIncrement()的實(shí)現(xiàn)舉例

AtomicInteger.java

 private static final Unsafe unsafe = Unsafe.getUnsafe();
public final int getAndIncrement() {
    for (;;) {
      int current = get();
      int next = current + 1;
      if (compareAndSet(current, next))
        return current;
    }
  }

public final boolean compareAndSet(int expect, int update) {
      return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    } 

留意這個(gè)for循環(huán)，只有在compareAndSet成功時(shí)才會(huì)返回。否則就一直compareAndSet。

調(diào)用了compareAndSet實(shí)現(xiàn)。此處，我注意到 Oracle JDK的實(shí)現(xiàn)是略有不同的，如果你查看JDK下的src，你可以看到Oracle JDK是調(diào)用的Unsafe的getAndIncrement()，但我相信Oracle JDK實(shí)現(xiàn)Unsafe.java的時(shí)候應(yīng)該也是只調(diào)用compareAndSet，因?yàn)橐粋€(gè)compareAndSet就可以實(shí)現(xiàn)增加、減少、設(shè)值的原子操作了。

Unsafe.java

 public native boolean compareAndSwapInt(Object obj, long offset,
                     int expect, int update); 

通過(guò)JNI調(diào)用的C++的實(shí)現(xiàn)。

natUnsafe.cc

 jboolean
sun::misc::Unsafe::compareAndSwapInt (jobject obj, jlong offset,
           jint expect, jint update)
{
 jint *addr = (jint *)((char *)obj + offset);
 return compareAndSwap (addr, expect, update);
}

static inline bool
compareAndSwap (volatile jint *addr, jint old, jint new_val)
{
 jboolean result = false;
 spinlock lock;
 if ((result = (*addr == old)))
  *addr = new_val;
 return result;
} 

Unsafe::compareAndSwapInt調(diào)用 static 函數(shù) compareAndSwap。而compareAndSwap又使用spinlock作為鎖。這里的spinlock有LockGuard的意味，構(gòu)造時(shí)加鎖，析構(gòu)時(shí)釋放。

我們需要聚焦在spinlock里。這里是保證spinlock釋放之前都是原子操作的真正實(shí)現(xiàn)。

什么是spinlock

spinlock，即自旋鎖，一種循環(huán)等待(busy waiting)以獲取資源的鎖。不同于mutex的阻塞當(dāng)前線程、釋放CPU資源以等待需求的資源，spinlock不會(huì)進(jìn)入掛起、等待條件滿(mǎn)足、重新競(jìng)爭(zhēng)CPU的過(guò)程。這意味著只有在 等待鎖的代價(jià)小于線程執(zhí)行上下文切換的代價(jià)時(shí),Spinlock才優(yōu)于mutex。

natUnsafe.cc

 class spinlock
{
 static volatile obj_addr_t lock;

public:

spinlock ()
 {
  while (! compare_and_swap (&lock, 0, 1))
   _Jv_ThreadYield ();
 }
 ~spinlock ()
 {
  release_set (&lock, 0);
 }
}; 

以一個(gè)靜態(tài)變量 static volatile obj_addr_t lock; 作為標(biāo)志位，通過(guò)C++　RAII實(shí)現(xiàn)一個(gè)Guard，所以所謂的鎖其實(shí)是 靜態(tài)成員變量obj_addr_t lock，C++中volatile 并不能保證同步，保證同步的是構(gòu)造函數(shù)里調(diào)用的 compare_and_swap和一個(gè)static變量lock.這個(gè)lock變量是1的時(shí)候，就需要等;是0的時(shí)候，就通過(guò)原子操作把它置為1，表示自己獲得了鎖。

這里會(huì)用一個(gè)static變量實(shí)在是一個(gè)意外，如此相當(dāng)于所有的無(wú)鎖結(jié)構(gòu)都共用同一個(gè)變量(實(shí)際就是size_t)來(lái)區(qū)分是否加鎖。當(dāng)這個(gè)變量置為1時(shí)，其他用到spinlock的都需要等。 為什么不在sun::misc::Unsafe添加一個(gè)私有變量 volatile obj_addr_t lock;，并作為構(gòu)造參數(shù)傳給spinlock？這樣相當(dāng)于每個(gè)UnSafe共享一個(gè)標(biāo)志位，效果會(huì)不會(huì)好一些？

_Jv_ThreadYield在下面的文件里，通過(guò)系統(tǒng)調(diào)用sched_yield(man 2 sched_yield)讓出CPU資源。宏HAVE_SCHED_YIELD在configure里定義，意味著編譯時(shí)如果取消定義，spinlock就稱(chēng)為真正意義的自旋鎖了。

posix-threads.h

 inline void
_Jv_ThreadYield (void)
{
#ifdef HAVE_SCHED_YIELD
 sched_yield ();
#endif /* HAVE_SCHED_YIELD */
}

這個(gè)lock.h在不同平臺(tái)有著不同的實(shí)現(xiàn)，我們以ia64(Intel AMD x64)平臺(tái)舉例，其他的實(shí)現(xiàn)可以在 這里 看到。

ia64/locks.h
 

typedef size_t obj_addr_t;
inline static bool
compare_and_swap(volatile obj_addr_t *addr,
             obj_addr_t old,
             obj_addr_t new_val)
{
 return __sync_bool_compare_and_swap (addr, old, new_val);
}

inline static void
release_set(volatile obj_addr_t *addr, obj_addr_t new_val)
{
 __asm__ __volatile__("" : : : "memory");
 *(addr) = new_val;
} 

__sync_bool_compare_and_swap 是gcc內(nèi)建函數(shù)，匯編指令"memory"完成內(nèi)存屏障。

1. 一般地，如果CPU硬件支持指令 cmpxchg (該指令從硬件保障原子性，毫無(wú)疑問(wèn)十分高效)，那么__sync_bool_compare_and_swap就應(yīng)該是用cmpxchg來(lái)實(shí)現(xiàn)的。
2. 不支持cmpxchg的CPU架構(gòu) 可以用lock指令前綴，通過(guò)鎖CPU總線的方式實(shí)現(xiàn)。
3. 如果連lock指令都不支持，有可能通過(guò)APIC實(shí)現(xiàn)
   

總之，硬件上保證多核CPU同步，而Unsafe的實(shí)現(xiàn)也是盡可能的高效。GCC-java的還算高效，相信Oracle 和 OpenJDK不會(huì)更差。

原子操作 和 GCC內(nèi)建的原子操作

原子操作

Java的表達(dá)式以及C++的表達(dá)式，都不是原子操作，也就是說(shuō) 你在代碼里：

 //假設(shè)i是線程間共享的變量
i++; 

在多線程環(huán)境下，i的訪問(wèn)是非原子性的，實(shí)際分成如下三個(gè)操作數(shù):

1. 從緩存取到寄存器
2. 在寄存器加1
3. 存入緩存
   

編譯器會(huì)改變執(zhí)行的時(shí)序，因此執(zhí)行結(jié)果可能并非所期望的。

GCC內(nèi)建的原子操作

gcc內(nèi)建了如下的原子操作，這些原子操作從4.1.2被加入。而之前，他們是使用內(nèi)聯(lián)的匯編實(shí)現(xiàn)的。

 type __sync_fetch_and_add (type *ptr, type value, ...)
type __sync_fetch_and_sub (type *ptr, type value, ...)
type __sync_fetch_and_or (type *ptr, type value, ...)
type __sync_fetch_and_and (type *ptr, type value, ...)
type __sync_fetch_and_xor (type *ptr, type value, ...)
type __sync_fetch_and_nand (type *ptr, type value, ...)

type __sync_add_and_fetch (type *ptr, type value, ...)
type __sync_sub_and_fetch (type *ptr, type value, ...)
type __sync_or_and_fetch (type *ptr, type value, ...)
type __sync_and_and_fetch (type *ptr, type value, ...)
type __sync_xor_and_fetch (type *ptr, type value, ...)
type __sync_nand_and_fetch (type *ptr, type value, ...)

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, ...)
type __sync_val_compare_and_swap (type *ptr, type oldval type newval, ...)

__sync_synchronize (...)

type __sync_lock_test_and_set (type *ptr, type value, ...)

void __sync_lock_release (type *ptr, ...) 

需要注意的是:

1.  __sync_fetch_and_add 和 __sync_add_and_fetch 的關(guān)系 對(duì)應(yīng)于 i++ 和 ++i。其他類(lèi)推
2.  CAS的兩種實(shí)現(xiàn)，bool版本的 如果對(duì)比oldval與ptr成功并給ptr設(shè)值newval 返回true；另一個(gè) 返回 原本*ptr的值
3.  __sync_synchronize 添加一個(gè)完全的內(nèi)存屏障
   

OpenJDK 的相關(guān)文件

下面列出一些Github上 OpenJDK9的原子操作實(shí)現(xiàn)，希望能幫助需要了解的人。畢竟OpenJDK比Gcc的實(shí)現(xiàn)應(yīng)用更廣泛一些。————但終究沒(méi)有Oracle JDK的源碼，雖然據(jù)說(shuō)OpenJDK與 Oracle的源碼差距很小。

AtomicInteger.java

Unsafe.java::compareAndExchangeObject

unsafe.cpp::Unsafe_CompareAndExchangeObject

oop.inline.hpp::oopDesc::atomic_compare_exchange_oop

atomic_linux_x86.hpp::Atomic::cmpxchg

 inline jlong  Atomic::cmpxchg  (jlong  exchange_value, volatile jlong*  dest, jlong  compare_value, cmpxchg_memory_order order) {
 bool mp = os::is_MP();
 __asm__ __volatile__ (LOCK_IF_MP(%4) "cmpxchgq %1,(%3)"
            : "=a" (exchange_value)
            : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
            : "cc", "memory");
 return exchange_value;
} 

這里需要給不熟悉C/C++的Java程序員提示一下，嵌入?yún)R編指令的格式如下

 __asm__ [__volatile__](assembly template//匯編模板
    : [output operand list]//輸入列表
    : [input operand list]//輸出列表
    : [clobber list])//破壞列表 

匯編模板中的%1,%3,%4對(duì)應(yīng)于后面的參數(shù)列表{"r" (exchange_value)，"r" (dest)，"r" (mp)}，參數(shù)列表以逗號(hào)分隔，從0排序。輸出參數(shù)放第一個(gè)冒號(hào)右邊，輸出參數(shù)放第二個(gè)冒號(hào)右邊。"r"表示放到通用寄存器，"a"表示寄存器EAX，有"="表示用于輸出(寫(xiě)還)。cmpxchg指令隱含使用EAX寄存器即參數(shù)%2.

其他細(xì)節(jié)就不在此羅列了，Gcc的實(shí)現(xiàn)是把要交換的指針傳下來(lái)，對(duì)比成功后直接賦值(賦值非原子)，原子性通過(guò)spinlock保證。 

OpenJDK的實(shí)現(xiàn)是把要交換的指針傳下來(lái)，直接通過(guò)匯編指令cmpxchgq賦值，原子性通過(guò)匯編指令保證。當(dāng)然gcc的spinlock底層也是通過(guò)cmpxchgq保證的。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。