簡介

• 多線程鎖定同一資源會造成死鎖
• 線程池中的任務使用當前線程池也可能出現(xiàn)死鎖
• RxJava 或 Reactor 等現(xiàn)代流行庫也可能出現(xiàn)死鎖

死鎖是兩個或多個線程互相等待對方所擁有的資源的情形。舉個例子，線程 A 等待 lock1，lock1 當前由線程 B 鎖住，然而線程 B 也在等待由線程 A 鎖住的 lock2。最壞情況下，應用程序將無限期凍結。讓我給你看個具體例子。假設這里有個?Lumberjack（伐木工） 類，包含了兩個裝備的鎖：

import com.google.common.collect.ImmutableList;
import lombok.RequiredArgsConstructor;
import java.util.concurrent.locks.Lock;
@RequiredArgsConstructor
class Lumberjack {
    private final String name;
    private final Lock accessoryOne;
    private final Lock accessoryTwo;
    void cut(Runnable work) {
        try {
            accessoryOne.lock();
            try {
                accessoryTwo.lock();
                work.run();
            } finally {
                accessoryTwo.unlock();
            }
        } finally {
            accessoryOne.unlock();
        }
    }
}

每個?Lumberjack（伐木工）需要兩件裝備：helmet（安全帽） 和?chainsaw（電鋸）。在他開始工作前，他必須擁有全部兩件裝備。我們通過如下方式創(chuàng)建伐木工們：

import lombok.RequiredArgsConstructor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@RequiredArgsConstructor
class Logging {
    private final Names names;
    private final Lock helmet = new ReentrantLock();
    private final Lock chainsaw = new ReentrantLock();
    Lumberjack careful() {
        return new Lumberjack(names.getRandomName(), helmet, chainsaw);
    }
    Lumberjack yolo() {
        return new Lumberjack(names.getRandomName(), chainsaw, helmet);
    }
}

可以看到，有兩種伐木工：先戴好安全帽然后再拿電鋸的，另一種則相反。謹慎派（careful()）伐木工先戴好安全帽，然后去拿電鋸�？褚芭煞ツ竟ぃ�yolo()）先拿電鋸，然后找安全帽。讓我們并發(fā)生成一些伐木工：

private List<Lumberjack> generate(int count, Supplier<Lumberjack> factory) {
    return IntStream
            .range(0, count)
            .mapToObj(x -> factory.get())
            .collect(toList());
}

generate()方法可以創(chuàng)建指定類型伐木工的集合。我們來生成一些謹慎派伐木工和狂野派伐木工。

private final Logging logging;
//...
List<Lumberjack> lumberjacks = new CopyOnWriteArrayList<>();
lumberjacks.addAll(generate(carefulLumberjacks, logging::careful));
lumberjacks.addAll(generate(yoloLumberjacks, logging::yolo));

最后，我們讓這些伐木工開始工作：

IntStream
        .range(0, howManyTrees)
        .forEach(x -> {
            Lumberjack roundRobinJack = lumberjacks.get(x % lumberjacks.size());
            pool.submit(() -> {
                log.debug("{} cuts down tree, {} left", roundRobinJack, latch.getCount());
                roundRobinJack.cut(/* ... */);
            });
        });

這個循環(huán)讓所有伐木工一個接一個（輪詢方式）去砍樹。實質(zhì)上，我們向線程池（ExecutorService）提交了和樹木數(shù)量（howManyTrees）相同個數(shù)的任務，并使用?CountDownLatch?來記錄工作是否完成。

CountDownLatch latch = new CountDownLatch(howManyTrees);
IntStream
        .range(0, howManyTrees)
        .forEach(x -> {
            pool.submit(() -> {
                //...
                roundRobinJack.cut(latch::countDown);
            });
        });
if (!latch.await(10, TimeUnit.SECONDS)) {
    throw new TimeoutException("Cutting forest for too long");
}

其實想法很簡單。我們讓多個伐木工（Lumberjacks）通過多線程方式去競爭一個安全帽和一把電鋸。完整代碼如下：

import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@RequiredArgsConstructor
class Forest implements AutoCloseable {
    private static final Logger log = LoggerFactory.getLogger(Forest.class);
    private final ExecutorService pool;
    private final Logging logging;
    void cutTrees(int howManyTrees, int carefulLumberjacks, int yoloLumberjacks) throws InterruptedException, TimeoutException {
        CountDownLatch latch = new CountDownLatch(howManyTrees);
        List<Lumberjack> lumberjacks = new ArrayList<>();
        lumberjacks.addAll(generate(carefulLumberjacks, logging::careful));
        lumberjacks.addAll(generate(yoloLumberjacks, logging::yolo));
        IntStream
                .range(0, howManyTrees)
                .forEach(x -> {
                    Lumberjack roundRobinJack = lumberjacks.get(x % lumberjacks.size());
                    pool.submit(() -> {
                        log.debug("{} cuts down tree, {} left", roundRobinJack, latch.getCount());
                        roundRobinJack.cut(latch::countDown);
                    });
                });
        if (!latch.await(10, TimeUnit.SECONDS)) {
            throw new TimeoutException("Cutting forest for too long");
        }
        log.debug("Cut all trees");
    }
    private List<Lumberjack> generate(int count, Supplier<Lumberjack> factory) {
        return IntStream
                .range(0, count)
                .mapToObj(x -> factory.get())
                .collect(Collectors.toList());
    }
    @Override
    public void close() {
        pool.shutdownNow();
    }
}

現(xiàn)在，讓我們來看有趣的部分。如果我們只創(chuàng)建謹慎派伐木工（careful Lumberjacks），應用程序幾乎瞬間運行完成，舉個例子：

ExecutorService pool = Executors.newFixedThreadPool(10);
Logging logging = new Logging(new Names());
try (Forest forest = new Forest(pool, logging)) {
    forest.cutTrees(10000, 10, 0);
} catch (TimeoutException e) {
    log.warn("Working for too long", e);
}

但是，如果你對伐木工（Lumberjacks）的數(shù)量做些修改，比如，10 個謹慎派（careful）伐木工和 1 個狂野派（yolo）伐木工，系統(tǒng)就會經(jīng)常運行失敗。怎么回事？謹慎派（careful）團隊里每個人都首先嘗試獲取安全帽。如果其中一個伐木工取到了安全帽，其他人會等待。然后那個幸運兒肯定能拿到電鋸。原因就是其他人在等待安全帽，還沒到獲取電鋸的階段。目前為止很完美。但是如果團隊里有一個狂野派（yolo）伐木工呢？當所有人競爭安全帽時，他偷偷把電鋸拿走了。這就出現(xiàn)問題了。某個謹慎派（careful）伐木工牢牢握著安全帽，但他拿不到電鋸，因為被其他某人拿走了。更糟糕的是電鋸所有者（那個狂野派伐木工）在拿到安全帽之前不會放棄電鋸。這里并沒有一個超時設定。那個謹慎派（careful）伐木工拿著安全帽無限等待電鋸，那個狂野派（yolo）伐木工因為拿不到安全帽也將永遠發(fā)呆，這就是死鎖。

如果所有伐木工都是狂野派（yolo）會怎樣，也就是說，所有人都首先去嘗試拿電鋸會怎樣？事實證明避免死鎖最簡單的方式就是以相同的順序獲取和釋放各個鎖，也就是說，你可以對你的資源按照某個標準來排序。如果一個線程先獲取 A 鎖，然后是 B 鎖，但第二個線程先獲取 B 鎖，會引發(fā)死鎖。

線程池自己引發(fā)的死鎖

這里有個與上面不同的死鎖例子，它證明了單個線程池使用不當時也會引發(fā)死鎖。假設你有一個?ExecutorService，和之前一樣，按照下面的方式運行。

ExecutorService pool = Executors.newFixedThreadPool(10);
pool.submit(() -> {
    try {
        log.info("First");
        pool.submit(() -> log.info("Second")).get();
        log.info("Third");
    } catch (InterruptedException | ExecutionException e) {
        log.error("Error", e);
    }
});

看起來沒什么問題 —— 所有信息按照預期的樣子呈現(xiàn)在屏幕上：

INFO [pool-1-thread-1]: First
INFO [pool-1-thread-2]: Second
INFO [pool-1-thread-1]: Third

注意我們用?get()?阻塞線程，在顯示“Third”之前必須等待內(nèi)部線程（Runnable）運行完成。這是個大坑！等待內(nèi)部任務完成意味著需要從線程池額外獲取一個線程來執(zhí)行任務。然而，我們已經(jīng)使用到了一個線程，所以內(nèi)部任務在獲取到第二個線程前將一直阻塞。當前我們的線程池足夠大，運行沒問題。讓我們稍微改變一下代碼，將線程池縮減到只有一個線程，另外關鍵的一點是我們移除?get()?方法：

ExecutorService pool = Executors.newSingleThreadExecutor();
pool.submit(() -> {
    log.info("First");
    pool.submit(() -> log.info("Second"));
    log.info("Third");
});

代碼正常運行，只是有些亂：

INFO [pool-1-thread-1]: First
INFO [pool-1-thread-1]: Third
INFO [pool-1-thread-1]: Second

兩點需要注意：

• 所有代碼運行在單個線程上（毫無疑問）
• “Third”信息顯示在“Second”之前

順序的改變完全在預料之內(nèi)，沒有涉及線程間的競態(tài)條件（事實上我們只有一個線程）。仔細分析一下發(fā)生了什么：我們向線程池提交了一個新任務（打印“Second”的任務），但這次我們不需要等待這個任務完成。因為線程池中唯一的線程被打印“First”和“Third”的任務占用，所以這個外層任務繼續(xù)執(zhí)行，并打印“Third”。當這個任務完成時，將單個線程釋放回線程池，內(nèi)部任務最終開始執(zhí)行，并打印“Second”。那么死鎖在哪里？來試試在內(nèi)部任務里加上?get()?方法：

ExecutorService pool = Executors.newSingleThreadExecutor();
pool.submit(() -> {
    try {
        log.info("First");
        pool.submit(() -> log.info("Second")).get();
        log.info("Third");
    } catch (InterruptedException | ExecutionException e) {
        log.error("Error", e);
    }
});

死鎖出現(xiàn)了！我們來一步一步分析：

• 打印“First”的任務被提交到只有一個線程的線程池
• 任務開始執(zhí)行并打印“First”
• 我們向線程池提交了一個內(nèi)部任務，來打印“Second”
• 內(nèi)部任務進入等待任務隊列。沒有可用線程因為唯一的線程正在被占用
• 我們阻塞住并等待內(nèi)部任務執(zhí)行結果。不幸的是，我們等待內(nèi)部任務的同時也在占用著唯一的可用線程
• get() 方法無限等待，無法獲取線程
• 死鎖

這是否意味單線程的線程池是不好的？并不是，相同的問題會在任意大小的線程池中出現(xiàn)，只不過是在高負載情況下才會出現(xiàn)，這維護起來更加困難。你在技術層面上可以使用一個無界線程池，但這樣太糟糕了。

Reactor/RxJava

請注意，這類問題也會出現(xiàn)在上層庫，比如?Reactor：

Scheduler pool = Schedulers.fromExecutor(Executors.newFixedThreadPool(10));
Mono
    .fromRunnable(() -> {
        log.info("First");
        Mono
                .fromRunnable(() -> log.info("Second"))
                .subscribeOn(pool)
                .block();  //VERY, VERY BAD!
        log.info("Third");
    })
    .subscribeOn(pool);

當你部署代碼，它似乎可以正常工作，但很不符合編程習慣。根源的問題是相通的，最后一行的?subscribeOn()?表示外層任務（Runnable）請求了線程池(pool)中一個線程，同時，內(nèi)部任務（Runnable）也試圖獲取一個線程。如果把基礎的線程池換成只包含單個線程的線程池，會發(fā)生死鎖。對于 RxJava/Reactor 來說，解決方案很簡單——用異步操作替代阻塞操作。

Mono
    .fromRunnable(() -> {
        log.info("First");
        log.info("Third");
    })
    .then(Mono
            .fromRunnable(() -> log.info("Second"))
            .subscribeOn(pool))
    .subscribeOn(pool)

防患于未然

并沒有徹底避免死鎖的方法。試圖解決問題的技術手段往往會帶來死鎖風險，比如共享資源和排它鎖。如果無法根治死鎖（或死鎖并不明顯，比如使用線程池），還是試著保證代碼質(zhì)量、監(jiān)控線程池和避免無限阻塞。我很難想象你情愿無限等待程序運行完成，如同?get()?方法和?block()?方法在沒有設定超時時間的情況下執(zhí)行。

感謝閱讀！

原文鏈接： dzone 翻譯： ImportNew.com - 一杯哈希不加鹽
譯文鏈接： http://www.importnew.com/30277.html
[ 轉載請保留原文出處、譯者和譯文鏈接。]

標簽： Google 安全 代碼

版權申明：本站文章部分自網(wǎng)絡，如有侵權，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉載文章言論不代表本站觀點！
本站所提供的圖片等素材，版權歸原作者所有，如需使用，請與原作者聯(lián)系。