摘要：在接下來的分鐘，你將會學(xué)會如何通過同步關(guān)鍵字，鎖和信號量來同步訪問共享可變變量。所以在使用樂觀鎖時，你需要每次在訪問任何共享可變變量之后都要檢查鎖，來確保讀鎖仍然有效。

原文：Java 8 Concurrency Tutorial: Synchronization and Locks
譯者：飛龍  
協(xié)議：CC BY-NC-SA 4.0

歡迎閱讀我的Java8并發(fā)教程的第二部分。這份指南將會以簡單易懂的代碼示例來教給你如何在Java8中進(jìn)行并發(fā)編程。這是一系列教程中的第二部分。在接下來的15分鐘，你將會學(xué)會如何通過同步關(guān)鍵字，鎖和信號量來同步訪問共享可變變量。

第一部分：線程和執(zhí)行器

第二部分：同步和鎖

第三部分：原子操作和 ConcurrentMap

這篇文章中展示的中心概念也適用于Java的舊版本，然而代碼示例適用于Java 8，并嚴(yán)重依賴于lambda表達(dá)式和新的并發(fā)特性。如果你還不熟悉lambda，我推薦你先閱讀我的Java 8 教程。

出于簡單的因素，這個教程的代碼示例使用了定義在這里的兩個輔助函數(shù)sleep(seconds) 和 stop(executor)。

在上一章中，我們學(xué)到了如何通過執(zhí)行器服務(wù)同時執(zhí)行代碼。當(dāng)我們編寫這種多線程代碼時，我們需要特別注意共享可變變量的并發(fā)訪問。假設(shè)我們打算增加某個可被多個線程同時訪問的整數(shù)。

我們定義了count字段，帶有increment()方法來使count加一：

int count = 0;

void increment() {
    count = count + 1;
}

當(dāng)多個線程并發(fā)調(diào)用這個方法時，我們就會遇到大麻煩：

ExecutorService executor = Executors.newFixedThreadPool(2);

IntStream.range(0, 10000)
    .forEach(i -> executor.submit(this::increment));

stop(executor);

System.out.println(count);  // 9965

我們沒有看到count為10000的結(jié)果，上面代碼的實(shí)際結(jié)果在每次執(zhí)行時都不同。原因是我們在不同的線程上共享可變變量，并且變量訪問沒有同步機(jī)制，這會產(chǎn)生競爭條件。

增加一個數(shù)值需要三個步驟：（1）讀取當(dāng)前值，（2）使這個值加一，（3）將新的值寫到變量。如果兩個線程同時執(zhí)行，就有可能出現(xiàn)兩個線程同時執(zhí)行步驟1，于是會讀到相同的當(dāng)前值。這會導(dǎo)致無效的寫入，所以實(shí)際的結(jié)果會偏小。上面的例子中，對count的非同步并發(fā)訪問丟失了35次增加操作，但是你在自己執(zhí)行代碼時會看到不同的結(jié)果。

幸運(yùn)的是，Java自從很久之前就通過synchronized關(guān)鍵字支持線程同步。我們可以使用synchronized來修復(fù)上面在增加count時的競爭條件。

synchronized void incrementSync() {
    count = count + 1;
}

在我們并發(fā)調(diào)用incrementSync()時，我們得到了count為10000的預(yù)期結(jié)果。沒有再出現(xiàn)任何競爭條件，并且結(jié)果在每次代碼執(zhí)行中都很穩(wěn)定：

ExecutorService executor = Executors.newFixedThreadPool(2);

IntStream.range(0, 10000)
    .forEach(i -> executor.submit(this::incrementSync));

stop(executor);

System.out.println(count);  // 10000

synchronized關(guān)鍵字也可用于語句塊：

void incrementSync() {
    synchronized (this) {
        count = count + 1;
    }
}

Java在內(nèi)部使用所謂的“監(jiān)視器”（monitor），也稱為監(jiān)視器鎖（monitor lock）或內(nèi)在鎖（ intrinsic lock）來管理同步。監(jiān)視器綁定在對象上，例如，當(dāng)使用同步方法時，每個方法都共享相應(yīng)對象的相同監(jiān)視器。

所有隱式的監(jiān)視器都實(shí)現(xiàn)了重入（reentrant）特性。重入的意思是鎖綁定在當(dāng)前線程上。線程可以安全地多次獲取相同的鎖，而不會產(chǎn)生死鎖（例如，同步方法調(diào)用相同對象的另一個同步方法）。

并發(fā)API支持多種顯式的鎖，它們由Lock接口規(guī)定，用于代替synchronized的隱式鎖。鎖對細(xì)粒度的控制支持多種方法，因此它們比隱式的監(jiān)視器具有更大的開銷。

鎖的多個實(shí)現(xiàn)在標(biāo)準(zhǔn)JDK中提供，它們會在下面的章節(jié)中展示。

ReentrantLock類是互斥鎖，與通過synchronized訪問的隱式監(jiān)視器具有相同行為，但是具有擴(kuò)展功能。就像它的名稱一樣，這個鎖實(shí)現(xiàn)了重入特性，就像隱式監(jiān)視器一樣。

讓我們看看使用ReentrantLock之后的上面的例子。

ReentrantLock lock = new ReentrantLock();
int count = 0;

void increment() {
    lock.lock();
    try {
        count++;
    } finally {
        lock.unlock();
    }
}

鎖可以通過lock()來獲取，通過unlock()來釋放。把你的代碼包裝在try-finally代碼塊中來確保異常情況下的解鎖非常重要。這個方法是線程安全的，就像同步副本那樣。如果另一個線程已經(jīng)拿到鎖了，再次調(diào)用lock()會阻塞當(dāng)前線程，直到鎖被釋放。在任意給定的時間內(nèi)，只有一個線程可以拿到鎖。

鎖對細(xì)粒度的控制支持多種方法，就像下面的例子那樣：

executor.submit(() -> {
    lock.lock();
    try {
        sleep(1);
    } finally {
        lock.unlock();
    }
});

executor.submit(() -> {
    System.out.println("Locked: " + lock.isLocked());
    System.out.println("Held by me: " + lock.isHeldByCurrentThread());
    boolean locked = lock.tryLock();
    System.out.println("Lock acquired: " + locked);
});

stop(executor);

在第一個任務(wù)拿到鎖的一秒之后，第二個任務(wù)獲得了鎖的當(dāng)前狀態(tài)的不同信息。

Locked: true
Held by me: false
Lock acquired: false

tryLock()方法是lock()方法的替代，它嘗試拿鎖而不阻塞當(dāng)前線程。在訪問任何共享可變變量之前，必須使用布爾值結(jié)果來檢查鎖是否已經(jīng)被獲取。

ReadWriteLock接口規(guī)定了鎖的另一種類型，包含用于讀寫訪問的一對鎖。讀寫鎖的理念是，只要沒有任何線程寫入變量，并發(fā)讀取可變變量通常是安全的。所以讀鎖可以同時被多個線程持有，只要沒有線程持有寫鎖。這樣可以提升性能和吞吐量，因?yàn)樽x取比寫入更加頻繁。

ExecutorService executor = Executors.newFixedThreadPool(2);
Map map = new HashMap<>();
ReadWriteLock lock = new ReentrantReadWriteLock();

executor.submit(() -> {
    lock.writeLock().lock();
    try {
        sleep(1);
        map.put("foo", "bar");
    } finally {
        lock.writeLock().unlock();
    }
});

上面的例子在暫停一秒之后，首先獲取寫鎖來向映射添加新的值。在這個任務(wù)完成之前，兩個其它的任務(wù)被啟動，嘗試讀取映射中的元素，并暫停一秒：

Runnable readTask = () -> {
    lock.readLock().lock();
    try {
        System.out.println(map.get("foo"));
        sleep(1);
    } finally {
        lock.readLock().unlock();
    }
};

executor.submit(readTask);
executor.submit(readTask);

stop(executor);

當(dāng)你執(zhí)行這一代碼示例時，你會注意到兩個讀任務(wù)需要等待寫任務(wù)完成。在釋放了寫鎖之后，兩個讀任務(wù)會同時執(zhí)行，并同時打印結(jié)果。它們不需要相互等待完成，因?yàn)樽x鎖可以安全同步獲取，只要沒有其它線程獲取了寫鎖。

Java 8 自帶了一種新的鎖，叫做StampedLock，它同樣支持讀寫鎖，就像上面的例子那樣。與ReadWriteLock不同的是，StampedLock的鎖方法會返回表示為long的標(biāo)記。你可以使用這些標(biāo)記來釋放鎖，或者檢查鎖是否有效。此外，StampedLock支持另一種叫做樂觀鎖（optimistic locking）的模式。

讓我們使用StampedLock代替ReadWriteLock重寫上面的例子：

ExecutorService executor = Executors.newFixedThreadPool(2);
Map map = new HashMap<>();
StampedLock lock = new StampedLock();

executor.submit(() -> {
    long stamp = lock.writeLock();
    try {
        sleep(1);
        map.put("foo", "bar");
    } finally {
        lock.unlockWrite(stamp);
    }
});

Runnable readTask = () -> {
    long stamp = lock.readLock();
    try {
        System.out.println(map.get("foo"));
        sleep(1);
    } finally {
        lock.unlockRead(stamp);
    }
};

executor.submit(readTask);
executor.submit(readTask);

stop(executor);

通過readLock() 或 writeLock()來獲取讀鎖或?qū)戞i會返回一個標(biāo)記，它可以在稍后用于在finally塊中解鎖。要記住StampedLock并沒有實(shí)現(xiàn)重入特性。每次調(diào)用加鎖都會返回一個新的標(biāo)記，并且在沒有可用的鎖時阻塞，即使相同線程已經(jīng)拿鎖了。所以你需要額外注意不要出現(xiàn)死鎖。

就像前面的ReadWriteLock例子那樣，兩個讀任務(wù)都需要等待寫鎖釋放。之后兩個讀任務(wù)同時向控制臺打印信息，因?yàn)槎鄠€讀操作不會相互阻塞，只要沒有線程拿到寫鎖。

下面的例子展示了樂觀鎖：

ExecutorService executor = Executors.newFixedThreadPool(2);
StampedLock lock = new StampedLock();

executor.submit(() -> {
    long stamp = lock.tryOptimisticRead();
    try {
        System.out.println("Optimistic Lock Valid: " + lock.validate(stamp));
        sleep(1);
        System.out.println("Optimistic Lock Valid: " + lock.validate(stamp));
        sleep(2);
        System.out.println("Optimistic Lock Valid: " + lock.validate(stamp));
    } finally {
        lock.unlock(stamp);
    }
});

executor.submit(() -> {
    long stamp = lock.writeLock();
    try {
        System.out.println("Write Lock acquired");
        sleep(2);
    } finally {
        lock.unlock(stamp);
        System.out.println("Write done");
    }
});

stop(executor);

樂觀的讀鎖通過調(diào)用tryOptimisticRead()獲取，它總是返回一個標(biāo)記而不阻塞當(dāng)前線程，無論鎖是否真正可用。如果已經(jīng)有寫鎖被拿到，返回的標(biāo)記等于0。你需要總是通過lock.validate(stamp)檢查標(biāo)記是否有效。

執(zhí)行上面的代碼會產(chǎn)生以下輸出：

Optimistic Lock Valid: true
Write Lock acquired
Optimistic Lock Valid: false
Write done
Optimistic Lock Valid: false

樂觀鎖在剛剛拿到鎖之后是有效的。和普通的讀鎖不同的是，樂觀鎖不阻止其他線程同時獲取寫鎖。在第一個線程暫停一秒之后，第二個線程拿到寫鎖而無需等待樂觀的讀鎖被釋放。此時，樂觀的讀鎖就不再有效了。甚至當(dāng)寫鎖釋放時，樂觀的讀鎖還處于無效狀態(tài)。

所以在使用樂觀鎖時，你需要每次在訪問任何共享可變變量之后都要檢查鎖，來確保讀鎖仍然有效。

有時，將讀鎖轉(zhuǎn)換為寫鎖而不用再次解鎖和加鎖十分實(shí)用。StampedLock為這種目的提供了tryConvertToWriteLock()方法，就像下面那樣：

ExecutorService executor = Executors.newFixedThreadPool(2);
StampedLock lock = new StampedLock();

executor.submit(() -> {
    long stamp = lock.readLock();
    try {
        if (count == 0) {
            stamp = lock.tryConvertToWriteLock(stamp);
            if (stamp == 0L) {
                System.out.println("Could not convert to write lock");
                stamp = lock.writeLock();
            }
            count = 23;
        }
        System.out.println(count);
    } finally {
        lock.unlock(stamp);
    }
});

stop(executor);

第一個任務(wù)獲取讀鎖，并向控制臺打印count字段的當(dāng)前值。但是如果當(dāng)前值是零，我們希望將其賦值為23。我們首先需要將讀鎖轉(zhuǎn)換為寫鎖，來避免打破其它線程潛在的并發(fā)訪問。tryConvertToWriteLock()的調(diào)用不會阻塞，但是可能會返回為零的標(biāo)記，表示當(dāng)前沒有可用的寫鎖。這種情況下，我們調(diào)用writeLock()來阻塞當(dāng)前線程，直到有可用的寫鎖。

除了鎖之外，并發(fā)API也支持計數(shù)的信號量。不過鎖通常用于變量或資源的互斥訪問，信號量可以維護(hù)整體的準(zhǔn)入許可。這在一些不同場景下，例如你需要限制你程序某個部分的并發(fā)訪問總數(shù)時非常實(shí)用。

下面是一個例子，演示了如何限制對通過sleep(5)模擬的長時間運(yùn)行任務(wù)的訪問：

ExecutorService executor = Executors.newFixedThreadPool(10);

Semaphore semaphore = new Semaphore(5);

Runnable longRunningTask = () -> {
    boolean permit = false;
    try {
        permit = semaphore.tryAcquire(1, TimeUnit.SECONDS);
        if (permit) {
            System.out.println("Semaphore acquired");
            sleep(5);
        } else {
            System.out.println("Could not acquire semaphore");
        }
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    } finally {
        if (permit) {
            semaphore.release();
        }
    }
}

IntStream.range(0, 10)
    .forEach(i -> executor.submit(longRunningTask));

stop(executor);

執(zhí)行器可能同時運(yùn)行10個任務(wù)，但是我們使用了大小為5的信號量，所以將并發(fā)訪問限制為5。使用try-finally代碼塊在異常情況中合理釋放信號量十分重要。

執(zhí)行上述代碼產(chǎn)生如下結(jié)果：

Semaphore acquired
Semaphore acquired
Semaphore acquired
Semaphore acquired
Semaphore acquired
Could not acquire semaphore
Could not acquire semaphore
Could not acquire semaphore
Could not acquire semaphore
Could not acquire semaphore

信號量限制對通過sleep(5)模擬的長時間運(yùn)行任務(wù)的訪問，最大5個線程。每個隨后的tryAcquire()調(diào)用在經(jīng)過最大為一秒的等待超時之后，會向控制臺打印不能獲取信號量的結(jié)果。

這就是我的系列并發(fā)教程的第二部分。以后會放出更多的部分，所以敬請等待吧。像以前一樣，你可以在Github上找到這篇文檔的所有示例代碼，所以請隨意fork這個倉庫，并自己嘗試它。

我希望你能喜歡這篇文章。如果你還有任何問題，在下面的評論中向我反饋。你也可以在Twitter上關(guān)注我來獲取更多開發(fā)相關(guān)的信息。

第一部分：線程和執(zhí)行器

第二部分：同步和鎖

第三部分：原子操作和 ConcurrentMap