這個專題我發現怎麼慢慢演化為效能測試了，遇到任何東西我就忍不住去測一把。本文我們會大概看一下各種鎖資料結構的簡單用法，順便也會來比拼一下效能。

各種併發鎖

首先，我們定一個抽象基類，用於各種鎖測試的一些公共程式碼：

• 我們需要使用鎖來保護counter和hashMap這2個資源
• write欄位表示這個執行緒是執行寫操作還是讀操作
• 每一個執行緒都會執行loopCount次讀或寫操作
• start的CountDownLatch用於等待所有執行緒一起執行
• finish的CountDownLatch用於讓主執行緒等待所有執行緒都完成

@Slf4j
abstract class LockTask implements Runnable {
    protected volatile static long counter;
    protected boolean write;
    protected static HashMap<Long, String> hashMap = new HashMap<>();
    int loopCount;
    CountDownLatch start;
    CountDownLatch finish;

    public LockTask(Boolean write) {
        this.write = write;
    }

    @Override
    public void run() {
        try {
            start.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < loopCount; i++) {
            doTask();
        }
        finish.countDown();
    }

    abstract protected void doTask();
}
 

下面我們實現最簡單的使用synchronized來實現的鎖，拿到鎖後我們針對hashMap和counter做一下最簡單的操作：

@Slf4j
class SyncTask extends LockTask {
    private static Object locker = new Object();

    public SyncTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        synchronized (locker) {
            if (write) {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } else {
                hashMap.get(counter);
                //log.debug("{}, {}", this.getClass().getSimpleName(), value);
            }
        }
    }
}
 

然後是ReentrantLock，使用也是很簡單，需要在finally中釋放鎖：

@Slf4j
class ReentrantLockTask extends LockTask {
    private static ReentrantLock locker = new ReentrantLock();

    public ReentrantLockTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        locker.lock();
        try {
            if (write) {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } else {
                hashMap.get(counter);
            }
        } finally {
            locker.unlock();
        }
    }
}
 

然後是ReentrantReadWriteLock，可重入的讀寫鎖，這屋裡我們需要區分讀操作還是寫操作來獲得不同型別的鎖：

@Slf4j
class ReentrantReadWriteLockTask extends LockTask {
    private static ReentrantReadWriteLock locker = new ReentrantReadWriteLock();

    public ReentrantReadWriteLockTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        if (write) {
            locker.writeLock().lock();
            try {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } finally {
                locker.writeLock().unlock();
            }
        } else {
            locker.readLock().lock();
            try {
                hashMap.get(counter);
            } finally {
                locker.readLock().unlock();
            }
        }
    }
}

然後是可重入鎖和可重入讀寫鎖的公平版本：

@Slf4j
class FairReentrantLockTask extends LockTask {
    private static ReentrantLock locker = new ReentrantLock(true);

    public FairReentrantLockTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        locker.lock();
        try {
            if (write) {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } else {
                hashMap.get(counter);
            }
        } finally {
            locker.unlock();
        }
    }
}

@Slf4j
class FairReentrantReadWriteLockTask extends LockTask {
    private static ReentrantReadWriteLock locker = new ReentrantReadWriteLock(true);

    public FairReentrantReadWriteLockTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        if (write) {
            locker.writeLock().lock();
            try {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } finally {
                locker.writeLock().unlock();
            }
        } else {
            locker.readLock().lock();
            try {
                hashMap.get(counter);
            } finally {
                locker.readLock().unlock();
            }
        }
    }
}

最後是1.8推出的StampedLock：

@Slf4j
class StampedLockTask extends LockTask {
    private static StampedLock locker = new StampedLock();

    public StampedLockTask(Boolean write) {
        super(write);
    }

    @Override
    protected void doTask() {
        if (write) {
            long stamp = locker.writeLock();
            try {
                counter++;
                hashMap.put(counter, "Data" + counter);
            } finally {
                locker.unlockWrite(stamp);
            }
        } else {
            long stamp = locker.tryOptimisticRead();
            long value = counter;

            if (!locker.validate(stamp)) {
                stamp = locker.readLock();
                try {
                    value = counter;
                } finally {
                    locker.unlockRead(stamp);
                }
            }
            hashMap.get(value);
        }
    }
}

這裡同樣區分讀寫鎖，只是讀鎖我們先嚐試進行樂觀讀，拿到一個戳後讀取我們需要保護的資料，隨後校驗一下這個戳如果沒問題的話說明資料沒有改變，樂觀鎖生效，如果有問題升級為悲觀鎖再讀取一次。因為StampedLock很複雜很容易用錯，真的打算用的話務必研讀官網的各種鎖升級的例子（樂觀讀到讀，樂觀讀到寫，讀到寫）。

效能測試和分析

同樣我們定義效能測試的型別：

@ToString
@RequiredArgsConstructor
class TestCase {
    final Class lockTaskClass;
    final int writerThreadCount;
    final int readerThreadCount;
    long duration;
}

每一種測試可以靈活選擇：

• 測試的鎖型別
• 寫執行緒數量
• 讀執行緒數量
• 最後測試結果回寫到duration

下面是效能測試的場景定義：

 @Test
public void test() throws Exception {
    List<TestCase> testCases = new ArrayList<>();

    Arrays.asList(SyncTask.class,
            ReentrantLockTask.class,
            FairReentrantLockTask.class,
            ReentrantReadWriteLockTask.class,
            FairReentrantReadWriteLockTask.class,
            StampedLockTask.class
    ).forEach(syncTaskClass -> {
        testCases.add(new TestCase(syncTaskClass, 1, 0));
        testCases.add(new TestCase(syncTaskClass, 10, 0));
        testCases.add(new TestCase(syncTaskClass, 0, 1));
        testCases.add(new TestCase(syncTaskClass, 0, 10));

        testCases.add(new TestCase(syncTaskClass, 1, 1));
        testCases.add(new TestCase(syncTaskClass, 10, 10));
        testCases.add(new TestCase(syncTaskClass, 50, 50));
        testCases.add(new TestCase(syncTaskClass, 100, 100));
        testCases.add(new TestCase(syncTaskClass, 500, 500));
        testCases.add(new TestCase(syncTaskClass, 1000, 1000));

        testCases.add(new TestCase(syncTaskClass, 1, 10));
        testCases.add(new TestCase(syncTaskClass, 10, 100));
        testCases.add(new TestCase(syncTaskClass, 10, 200));
        testCases.add(new TestCase(syncTaskClass, 10, 500));
        testCases.add(new TestCase(syncTaskClass, 10, 1000));

        testCases.add(new TestCase(syncTaskClass, 10, 1));
        testCases.add(new TestCase(syncTaskClass, 100, 10));
        testCases.add(new TestCase(syncTaskClass, 200, 10));
        testCases.add(new TestCase(syncTaskClass, 500, 10));
        testCases.add(new TestCase(syncTaskClass, 1000, 10));

    });

    testCases.forEach(testCase -> {
        System.gc();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            benchmark(testCase);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });

    StringBuilder stringBuilder = new StringBuilder();
    int index = 0;
    for (TestCase testCase : testCases) {
        if (index % 20 == 0)
            stringBuilder.append("\r\n");
        stringBuilder.append(testCase.duration);
        stringBuilder.append(",");
        index++;
    }
    System.out.println(stringBuilder.toString());
}

在這裡可以看到，我們為這6個鎖定義了20種測試場景，覆蓋幾大類：

• 只有讀的情況
• 只有寫的情況
• 讀寫併發的情況，併發數漸漸增多
• 讀比寫多的情況（這個最常見吧）
• 寫比讀多的情況

每一次測試之間強制觸發gc後休眠1秒，每20次結果換行一次輸出。
測試類如下：

private void benchmark(TestCase testCase) throws Exception {
    LockTask.counter = 0;
    log.info("Start benchmark:{}", testCase);
    CountDownLatch start = new CountDownLatch(1);
    CountDownLatch finish = new CountDownLatch(testCase.readerThreadCount + testCase.writerThreadCount);
    if (testCase.readerThreadCount > 0) {
        LockTask readerTask = (LockTask) testCase.lockTaskClass.getDeclaredConstructor(Boolean.class).newInstance(false);
        readerTask.start = start;
        readerTask.finish = finish;
        readerTask.loopCount = LOOP_COUNT / testCase.readerThreadCount;
        if (testCase.lockTaskClass.getSimpleName().startsWith("Fair")) readerTask.loopCount /= 100;
        IntStream.rangeClosed(1, testCase.readerThreadCount)
                .mapToObj(__ -> new Thread(readerTask))
                .forEach(Thread::start);
    }
    if (testCase.writerThreadCount > 0) {
        LockTask writerTask = (LockTask) testCase.lockTaskClass.getDeclaredConstructor(Boolean.class).newInstance(true);
        writerTask.start = start;
        writerTask.finish = finish;
        writerTask.loopCount = LOOP_COUNT / testCase.writerThreadCount;
        if (testCase.lockTaskClass.getSimpleName().startsWith("Fair")) writerTask.loopCount /= 100;
        IntStream.rangeClosed(1, testCase.writerThreadCount)
                .mapToObj(__ -> new Thread(writerTask))
                .forEach(Thread::start);
    }

    start.countDown();
    long begin = System.currentTimeMillis();
    finish.await();
    if (testCase.writerThreadCount > 0) {
        if (testCase.lockTaskClass.getSimpleName().startsWith("Fair")) {
            Assert.assertEquals(LOOP_COUNT / 100, LockTask.counter);
        } else {
            Assert.assertEquals(LOOP_COUNT, LockTask.counter);
        }
    }
    testCase.duration = System.currentTimeMillis() - begin;
    log.info("Finish benchmark:{}", testCase);
}

程式碼主要乾了幾件事情：

• 根據測試用例的讀寫執行緒數，開啟一定量的執行緒，根據類名和讀寫型別動態建立型別
• 每一個執行緒執行的迴圈次數是按比例均勻分配的，公平型別的兩次測試數/100，因為實在是太慢了，等不了幾小時
• 使用兩個CountDownLatch來控制所有執行緒開啟，等待所有執行緒完成，最後校驗一下counter的總數

在這裡，我們把迴圈次數設定為1000萬次，在阿里雲12核12G機器JDK8環境下執行得到的結果如下：

這裡，我們進行兩次測試，其實一開始我的測試程式碼裡沒有HashMap的讀寫操作，只有counter的讀寫操作（這個時候迴圈次數是1億次），所有第一次測試是僅僅只有counter的讀寫操作的，後一次測試是這裡貼的程式碼的版本。

所以這個表格中的資料不能直接來對比因為混雜了三種迴圈次數，上面那個表是1億從迴圈的時間，下面那個是1000萬次，黃色的兩條分別是100萬次和10萬次迴圈。

這個測試資訊量很大，這裡說一下我看到的幾個結論，或者你還可以從這個測試中品味出其它結論：

• synchronized關鍵字經過各種優化進行簡單鎖的操作效能已經相當好了，如果用不到ReentrantLock高階功能的話，使用synchronized不會有什麼太多效能問題
• 在任務非常輕的時候可重入鎖比synchronized還是快那麼一點，一般場景下不可能只是++操作，這個時候兩者差不多
• 併發上來之後各種鎖的執行耗時稍微增多點，沒有增多太厲害，併發不足的時候反而效能還不好
• 在任務很輕的時候StampedLock效能碾壓群雄，在只有讀操作的時候因為只是樂觀鎖，所以效能好的誇張
• 在任務沒有那麼輕的時候讀寫鎖的效能幾乎都比普通鎖好，看下面那個表格，在任務實在是太輕的時候讀寫鎖因為複雜的鎖實現開銷的問題不如普通的可重入鎖
• 公平版本的鎖非常非常慢，可以說比非公平版本的慢100倍還不止，而且執行的時候CPU打滿，其它版本的鎖執行的時候CPU利用在12核的20%左右，其實想想也對，不管是多少執行緒，大部分時候都阻塞了

所以說對於這些鎖的選擇也很明確：

• 如果用不到ReentrantLock的什麼高階特性，synchronized就可以
• 一般而言ReentrantLock完全可以替代synchronized，如果你不嫌麻煩的話
• ReentrantReadWriteLock用於相對比較複雜的任務的讀寫併發的情況
• StampedLock用於相對比較輕量級任務的高併發的情況，用起來也比較複雜，能夠實現極致的效能
• 只有有特殊需求的話才去開啟ReentrantLock或ReentrantReadWriteLock的公平特性

再來看看ReentrantLock

之前也提到了可重入鎖相對synchronized有一些高階特性，我們寫一些測試程式碼：

• 我們先在主執行緒鎖10次
• 輸出一下鎖的一些資訊
• 迴圈10次開啟10個執行緒嘗試獲取鎖，等待時間是1秒到10秒，顯然主執行緒釋放鎖之前是獲取不到鎖的
• 1秒一次定時輸出鎖的一些資訊
• 5秒後主執行緒釋放鎖
• 休眠一下觀察子執行緒是否拿到鎖了

@Test
public void test() throws InterruptedException {

    ReentrantLock reentrantLock = new ReentrantLock(true);
    IntStream.rangeClosed(1, 10).forEach(i -> reentrantLock.lock());
    log.info("getHoldCount:{},isHeldByCurrentThread:{},isLocked:{}",
            reentrantLock.getHoldCount(),
            reentrantLock.isHeldByCurrentThread(),
            reentrantLock.isLocked());

    List<Thread> threads = IntStream.rangeClosed(1, 10).mapToObj(i -> new Thread(() -> {
        try {
            if (reentrantLock.tryLock(i, TimeUnit.SECONDS)) {
                try {
                    log.debug("Got lock");
                } finally {
                    reentrantLock.unlock();
                }
            } else {
                log.debug("Cannot get lock");
            }
        } catch (InterruptedException e) {
            log.debug("InterruptedException Cannot get lock");
            e.printStackTrace();
        }
    })).collect(Collectors.toList());

    Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> log.info("getHoldCount:{}, getQueueLength:{}, hasQueuedThreads:{}, waitThreads:{}",
            reentrantLock.getHoldCount(),
            reentrantLock.getQueueLength(),
            reentrantLock.hasQueuedThreads(),
            threads.stream().filter(reentrantLock::hasQueuedThread).count()), 0, 1, TimeUnit.SECONDS);

    threads.forEach(Thread::start);

    TimeUnit.SECONDS.sleep(5);
    IntStream.rangeClosed(1, 10).forEach(i -> reentrantLock.unlock());
    TimeUnit.SECONDS.sleep(1);
}

輸出如下：

08:14:50.834 [main] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:10,isHeldByCurrentThread:true,isLocked:true
08:14:50.849 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:10, hasQueuedThreads:true, waitThreads:10
08:14:51.849 [Thread-0] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Cannot get lock
08:14:51.848 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:9, hasQueuedThreads:true, waitThreads:9
08:14:52.849 [Thread-1] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Cannot get lock
08:14:52.849 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:8, hasQueuedThreads:true, waitThreads:8
08:14:53.846 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:8, hasQueuedThreads:true, waitThreads:8
08:14:53.847 [Thread-2] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Cannot get lock
08:14:54.847 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:7, hasQueuedThreads:true, waitThreads:7
08:14:54.849 [Thread-3] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Cannot get lock
08:14:55.847 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:6, hasQueuedThreads:true, waitThreads:6
08:14:55.850 [Thread-4] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Cannot get lock
08:14:55.850 [Thread-5] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Got lock
08:14:55.851 [Thread-6] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Got lock
08:14:55.852 [Thread-7] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Got lock
08:14:55.852 [Thread-8] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Got lock
08:14:55.852 [Thread-9] DEBUG me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - Got lock
08:14:56.849 [pool-1-thread-1] INFO me.josephzhu.javaconcurrenttest.lock.ReentrantLockTest - getHoldCount:0, getQueueLength:0, hasQueuedThreads:false, waitThreads:0

從這個輸出可以看到：

• 一開始顯示鎖被主執行緒鎖了10次
• 隨著時間的推移等待鎖的執行緒數量在增加
• 5個執行緒因為超時無法獲取到鎖
• 5秒後還有5個執行緒拿到了鎖

這也可以看到可重入鎖相比synchronized功能更強大點：

• 可以超時等待獲取鎖
• 可以檢視到鎖的一些資訊
• 可以中斷鎖（這裡沒有演示）
• 之前提到的公平性
• 可重入特性並不是它特有的功能，synchronized也能重入

提到了可重入，我們進行一個無聊的實驗看看可以重入多少次：

@Test
public void test2() {
    ReentrantLock reentrantLock = new ReentrantLock(true);
    int i = 0;
    try {
        while (true) {
            reentrantLock.lock();
            i++;
        }
    } catch (Error error) {
        log.error("count:{}", i, error);
    }
}

結果如下：

鎖誤用的例子

最後再提下最簡單的鎖誤用的例子，雖然沒有那麼高大上，但是這種因為鎖範圍和鎖保護物件的範圍不一致導致誤用的問題在業務程式碼中到處都是，比如：

@Slf4j
public class LockMisuse {

    @Test
    public void test1() throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        IntStream.rangeClosed(1, 100000).forEach(i -> executorService.submit(new Container()::test));
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.HOURS);
        log.info("{}", Container.counter);
    }
}

class Container {
    static int counter = 0;
    Object locker = new Object();

    void test() {
        synchronized (locker) {
            counter++;
        }
    }
}

在程式碼裡我們要保護的資源是靜態的，但是鎖卻是物件級別的，不同的例項持有不同的鎖，完全起不到保護作用：

小結

本文我們簡單測試了一下各種鎖的效能，我感覺這個測試可能還無法100%模擬真實的場景，真實情況下不僅僅是讀寫執行緒數量的不一致，更多是操作頻次的不一致，不過這個測試基本看到了我們猜測的結果。在日常程式碼開發過程中，大家可以根據實際功能和場景需要來選擇合適的鎖型別。

有的時候高大上的一些鎖因為使用複雜容易導致誤用、錯用、死鎖、活鎖等問題，我反而建議在沒有明顯問題的情況下先從簡單的『悲觀』鎖開始使用。還有就是像最後的例子，使用鎖的話務必需要認證檢查程式碼，思考鎖和保護物件的關係，避免鎖不產產生效果導致隱藏的Bug。

同樣，程式碼見我的Github，歡迎clone後自己把玩，歡迎點贊。

歡迎關注我的微信公眾號：隨緣主人的園子

相關推薦