Java 8 并發(fā)：同步和鎖

andycall 發(fā)布于2019-08-16 11:00 / 1024人閱讀

摘要：可重入意味著鎖被綁定到當前線程，線程可以安全地多次獲取相同的鎖，而不會發(fā)生死鎖例如同步方法在同一對象上調(diào)用另一個同步方法。寫入鎖釋放后，兩個任務(wù)并行執(zhí)行，它們不必等待對方是否完成，因為只要沒有線程持有寫入鎖，它們就可以同時持有讀取鎖。

原文地址： Java 8 Concurrency Tutorial: Synchronization and Locks

為了簡單起見，本教程的示例代碼使用了在這里定義的兩個輔助方法，sleep(seconds) 和 stop(executor)

Synchronized

當我們編寫多線程代碼訪問可共享的變量時需要特別注意，下面是一個多線程去改變一個整數(shù)的例子。

定義一個變量 count，定義一個方法 increment() 使 count 增加 1.

int count = 0;

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

當多個線程同時調(diào)用 increment() 時就會出現(xiàn)問題：

ExecutorService executor = Executors.newFixedThreadPool(2);

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

stop(executor);

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

上面的代碼執(zhí)行結(jié)果并不是10000，原因是我們在不同的線程上共享一個變量，而沒有給這個變量的訪問設(shè)置競爭條件。

為了增加數(shù)字，必須執(zhí)行三個步驟：(i) 讀取當前值；(ii) 將該值增加1；(iii) 將新值寫入變量；如果兩個線程并行執(zhí)行這些步驟，則兩個線程可能同時執(zhí)行步驟1，從而讀取相同的當前值。這導(dǎo)致寫入丟失，所以實際結(jié)果較低。在上面的示例中，35個增量由于并發(fā)非同步訪問計數(shù)而丟失，但是當你自己執(zhí)行代碼時可能會看到不同的結(jié)果。

幸運的是，Java 早期通過 synchronized 關(guān)鍵字支持線程同步。增加計數(shù)時，我們可以利用同步來解決上述競爭條件：

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

當我們使用 incrementSync() 方法時，我們得到了希望的結(jié)果，而且每次執(zhí)行的結(jié)果都是這樣的。

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;
    }
}

在 JVM 的內(nèi)部使用了一個監(jiān)視器，也可以稱為監(jiān)視器鎖和內(nèi)部鎖來管理同步。這個監(jiān)視器被綁定到一個對象上，當使用同步方法時，每個方法共享相應(yīng)對象的監(jiān)視器。

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

Locks

除了使用關(guān)鍵字 synchronized 支持的隱式鎖(對象的內(nèi)置鎖)外，Concurrency API 支持由 Lock 接口指定的各種顯示鎖。顯示鎖能控制更細的粒度，因此也有更好的性能，在邏輯上也比較清晰。

標準 JDK中提供了多種顯示鎖的實現(xiàn)，將在下面的章節(jié)中進行介紹。

ReentrantLock

ReentrantLock 類是一個互斥鎖，它和 synchronized 關(guān)鍵字訪問的隱式鎖具有相同的功能，但它具有擴展功能。它也實現(xiàn)了可重入的功能。

下面來看看如何使用 ReentrantLock

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

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

鎖通過 lock() 獲取，通過 unlock() 釋放，將代碼封裝到 try/finally 塊中是非常重要的，以確保在出現(xiàn)異常的時候也能釋放鎖。這個方法和使用關(guān)鍵字 synchronized 修飾的方法是一樣是線程安全的。如果一個線程已經(jīng)獲得了鎖，后續(xù)線程調(diào)用 lock() 會暫停線程，直到鎖被釋放，永遠只有一個線程能獲取鎖。

lock 支持更細粒度的去控制一個方法的同步，如下面的代碼：

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

executor.submit(() -> {
    lock.lock();
    try {
        sleep(1000);
    } 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);

當?shù)谝粋€任務(wù)獲取鎖時，第二個任務(wù)獲取鎖的狀態(tài)信息：

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

作為 lock() 方法的替代方法 tryLock() 嘗試去獲取鎖而不暫停當前線程，必須使用 bool 結(jié)果去判斷是否真的獲取到了鎖。

ReadWriteLock

ReadWriteLock 指定了另一種類型的鎖，即讀寫鎖。讀寫鎖實現(xiàn)的邏輯是，當沒有線程在寫這個變量時，其他的線程可以讀取這個變量，所以就是當沒有線程持有寫鎖時，讀鎖就可以被所有的線程持有。如果讀取比寫更頻繁，這將增加系統(tǒng)的性能和吞吐量。

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

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

上面的例子首先獲取一個寫入鎖，在 sleep 1秒后在 map 中寫入值，在這個任務(wù)完成之前，還有兩個任務(wù)正在提交，試圖從 map 讀取值：

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

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

stop(executor);

當執(zhí)行上面的代碼時，你會注意到兩人讀取的任務(wù)必須等待直到寫入完成(當在讀取的時候，寫是不能獲取鎖的)。寫入鎖釋放后，兩個任務(wù)并行執(zhí)行，它們不必等待對方是否完成，因為只要沒有線程持有寫入鎖，它們就可以同時持有讀取鎖。

StampedLock

Java 8 提供了一種新類型的鎖 StampedLock，像上面的例子一樣它也支持讀寫鎖，與 ReadWriteLock 不同的是，StampedLock 的鎖定方法返回一個 long 值，可以利用這個值檢查是否釋放鎖和鎖仍然有效。另外 StampedLock 支持另外一種稱為樂觀鎖的模式。

下面使用 StampedLock 來替換 ReadWriteLock

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

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

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

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

stop(executor);

通過 readLock() 和 writeLock() 方法來獲取讀寫鎖會返回一個稍后用于在 finally 塊中釋放鎖的值。注意，這里的鎖不是可重入的。每次鎖定都會返回一個新的值，并在沒有鎖的情況下阻塞，在使用的時候要注意不要死鎖。

就像前面 ReadWriteLock 中的示例一樣，兩個讀取任務(wù)必須等待寫入任務(wù)釋放鎖。然后同時并行執(zhí)行打印結(jié)果到控制臺。

下面的例子演示了樂觀鎖

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(1000);
        System.out.println("Optimistic Lock Valid: " + lock.validate(stamp));
        sleep(2000);
        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(2000);
    } finally {
        lock.unlock(stamp);
        System.out.println("Write done");
    }
});

stop(executor);

通過調(diào)用 tryOptimisticRead() 來獲取樂觀讀寫鎖，tryOptimisticRead()總是返回一個值，而不會阻塞當前線程，也不關(guān)鎖是否可用。如果有一個寫鎖激活則返回0。可以通過 lock.validate(stamp) 來檢查返回的標記(long 值)是否有效。

執(zhí)行上面的代碼輸出：

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

樂觀鎖在獲得鎖后立即生效。與普通讀鎖相反，樂觀鎖不會阻止其他線程立即獲得寫鎖。在第一個線程休眠一秒之后，第二個線程獲得一個寫鎖，而不用等待樂觀讀鎖解除。樂觀的讀鎖不再有效，即使寫入鎖定被釋放，樂觀的讀取鎖仍然無效。

因此，在使用樂觀鎖時，必須在每次訪問任何共享的變量后驗證鎖，以確保讀取仍然有效。

有時將讀鎖轉(zhuǎn)換為寫鎖并不需要再次解鎖和鎖定是有用的。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ù)首先獲得一個讀鎖，并將當前的變量計數(shù)值打印到控制臺。但是，如果當前值為 0，我們要分配一個新的值23。我們首先必須將讀鎖轉(zhuǎn)換為寫鎖，以不打破其他線程的潛在并發(fā)訪問。調(diào)用 tryConvertToWriteLock() 不會阻塞，但可能會返回 0，指示當前沒有寫鎖定可用。在這種情況下，我們調(diào)用writeLock（）來阻塞當前線程，直到寫鎖可用。

Semaphores

除了鎖之外，并發(fā)API還支持計數(shù)信號量。鎖通常授予對變量或資源的獨占訪問權(quán)，而信號量則能夠維護整套許可證。在不同的情況下，必須限制對應(yīng)用程序某些部分的并發(fā)訪問量。

下面是一個如何限制對長時間任務(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(5000);
        } 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í)行程序可以同時運行10個任務(wù)，但是我們使用5信號量，因此限制并發(fā)訪問為5個。使用try/finally塊，即使在異常的情況下正確釋放信號量也是非常重要的。

運行上面的代碼輸出：

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

當有 5 個任務(wù)獲取型號量后，隨后的任務(wù)便不能獲取信號量了。但是如果前面 5 的任務(wù)執(zhí)行完成，finally 塊釋放了型號量，隨后的線程就可以獲取星號量了，總數(shù)不會超過5個。這里調(diào)用 tryAcquire() 獲取型號量設(shè)置了超時時間1秒，意味著當線程獲取信號量失敗后可以阻塞等待1秒再獲取。