透過 Rust 探索系統的本原：併發原語

幾周前我寫了篇關於併發的文章（透過 rust 探索系統的本原：併發篇），從使用者的角度介紹了常用的處理併發的工具：Mutex / RwLock / Channel，以及 async/await。今天我們講講這些併發手段背後的原語。這些原語，大家在操作系統課程時大多學過，但如果不是做一些底層的開發，估計大家都不記得了。今天，我們就來簡單聊聊這些基礎的併發原語，瞭解它們的差異，明白它們使用的場景，對撰寫高性能的併發應用有很大的幫助。

有同學可能會問：我一個寫 web 的，需要 synchronize 的時候靠 db / message queue，再不濟用 Redlock [1]，瞭解這些玩意兒有啥用？嗯，有點道理。如果你的工作大部分是 CRUD，寫一些和數據庫打交道的 HTTP API，這些東西的確用處不大，也許你不是本文的讀者。但如果你想讓自己的技能樹稍微豐富一些，能做一些別人做不了的事情，能面對不同的場景設計出來更高效的系統，那麼這篇文章也許值得一讀。

今天我們講的東西，可能會略微枯燥，略微難懂，我會盡量引入足夠的擴展知識，把上下文講清楚。我們重點講解和深入幾個概念：atomic，mutex，condvar 和 channel。

Atomic

Atomic 是所有併發原語的基礎。在具體介紹 atomic 之前，我們先考慮一下，最基本的鎖該如何實現。我們假設要用一把鎖來保護某個數據結構的修改，使其在多線程環境下可以正常工作（獨佔或者互斥訪問）。爲了簡便起見，我們在獲取這把鎖的時候，如果獲取不到，就一直死循環，直到拿到鎖爲止：

struct Lock<T> {
  locked: bool,
  data: T,
}
impl Lock<T> {
  pub fn lock<R>(&mut self, op: impl FnOnce(&mut T) -> R) -> R {
    // spin if we can't get lock
    while self.locked != false {} // **1
    // ha, we can lock and do our job
    self.locked = true; // **2
    // execute the op as we got lock
    let result = op(self.data); // **3
    // unlock
    self.locked = false; // **4
    result
  }
}

這段代碼是編譯不過的，因爲按照 Rust 的借用規則，T 不能安全地在多個線程間存在可變引用。爲了避免非 Rust 背景的同學看得太暈，我省去了一些代碼 [2]，因爲我們的關注點是 lock 的實現本身。

這樣一個實現看上去似乎問題不大，但它有好幾個問題：

在多核情況下，**1 和 **2 之間，有可能其它線程也碰巧 spin 結束，把 locked 修改爲 true。這樣，存在多個線程拿到這把鎖，從而破壞了任何線程都有獨佔訪問的保證。
即便在單核情況下，**1 和 **2 之間，也可能因爲操作系統的可搶佔式調度，導致上述情況發生。
如今的編譯器會最大程度優化生成的指令 —— 如果操作之間沒有依賴關係，那麼可能會生成亂序的機器碼，比如：**3 被優化放在 **1 之前，從而破壞了這個 lock 的保證。
即便編譯器不做亂序處理，CPU 也會最大程度做指令的亂序執行，讓流水線的效率最高。同樣會發生 3 中的問題。

所以，我們實現的這個鎖的行爲是未定義的。可能大部分時間如你所願，但隨機出現奇奇怪怪的行爲。一旦這樣的事情發生，那麼，bug 可能會以各種不同的面貌出現在系統的各個角落。而且，這樣的 bug 幾乎是無解的：它很難穩定復現，表現行爲很不一致，甚至，只在某個 CPU 下出現。

爲了解決這樣的問題，我們必須在 CPU 層面做一些保證，讓某些操作成爲原子操作，其中最基礎的保證是：可以通過一條指令讀取某個內存地址，判斷其值是否等於某個前置值，如果相等，將其修改爲新的值。這就是 Compare-and-swap 操作，簡稱 CAS [3]。這個操作是操作系統的幾乎所有併發原語的基石，它使得我們可以實現一個可以正常工作的鎖。

對於上述的代碼，我們可以把一開始的循環改成：

while self
  .locked
  .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
  .is_error() {}

這句的意思是：如果 locked 當前的值是 false，那麼就將其改成 true。整個這個操作在一條指令裏完成，不會被其它線程打斷或者修改；如果 locked 的當前值不是 false，那麼就會返回錯誤，我們會在此不停 spin，直到前置條件得到滿足。這裏，compare_exchange 是 Rust 提供的 CAS 操作，它會被編譯成 CPU 的對應的 CAS 指令。

當這句執行成功後，locked 必然會被改變爲 true，我們成功拿到了鎖，而任何其他線程都會在這句話上 spin。

在釋放鎖的時候，我們相應地需要使用 atomic 的版本，而非直接賦值成 false：

self.locked.store(false, Ordering::Release);

當然，爲了配合這樣的改動，我們還需要把 locked 從 bool 改成 AtomicBool。在 Rust 裏，std::sync::atomic 有大量的 atomic 數據結構，他們對應了各種基礎結構。

通過使用 compare_exchange 我們可以規避上面 1 和 2 面臨的問題，但對於 3 和 4，我們還需要一些額外處理。這就是這個函數里額外的兩個和 Ordering 有關的奇怪的參數。

如果你查看 atomic 的文檔 [4]，可以看到 Ordering 是一個 enum：

pub enum Ordering {
    Relaxed,
    Release,
    Acquire,
    AcqRel,
    SeqCst,
}

文檔裏解釋了幾種 Ordering 的用途，我來稍稍擴展一下：

Relaxed：這是最寬鬆的規則，它對編譯器和 CPU 不做任何限制，可以亂序
Release：當我們寫入數據（上面的 store）的時候，如果用了 Release order，那麼：

對於其它線程，如果它們使用了 Acquire 來讀取這個 atomic 的數據，那麼它們看到的是修改後的結果。因爲上文中我們在 compare_exchange 裏使用了 Acquire 來讀取，所以保證讀到最新的值。
對於當前線程，任何讀取或寫入操作都不能被被亂序排在這個 store 之後。也就是說，在上文的例子裏，CPU 或者編譯器不能把 **3 挪到 **4 之後執行。

Acquire：當我們讀取數據的時候，如果用了 Acquire order，那麼：

對於其它線程，如果使用了 Release 來修改數據，那麼，修改的值對當前線程可見。
對於當前線程，任何讀取或者寫入操作都不能被亂序排在這個讀取之前。在上文的例子裏，CPU 或者編譯器不能把 **3 挪到 **1 之前執行。

AcqRel：Acquire 和 Release 的結合，同時擁有 Acquire 和 Release 的保證。這個一般用在 fetch_xxx 上，比如你要對一個 atomic 自增 1，你希望這個操作之前和之後的讀取或寫入操作不會被亂序，並且操作的結果對其它線程可見。
SeqCst：最嚴格的 ordering，除了 AcqRel 的保證外，它還保證所有線程看到的所有的 SeqCst 操作的順序是一致的。

因爲 CAS 和 ordering 都是系統級的操作，所以上面我描述的 Ordering 的用途在各種語言中都大同小異。對於 Rust 來說，它的 atomic 原語是繼承於 C++，見 [5]。如果讀 Rust 的文檔你感覺雲裏霧裏，那麼 C++ 的關於 ordering 的文檔要清晰得多。

好，上述的鎖的實現的完整代碼如下：

pub fn with_lock<R>(&self, op: impl FnOnce(&mut T) -> R) -> R {
    while self
        .locked
        .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
        .is_err()
    {
        while self.locked.load(Ordering::Relaxed) == true {}
    }
    let ret = op(unsafe { &mut *self.v.get() });
    self.locked.store(false, Ordering::Release);
    ret
}

注意，我們在 while loop 裏，又嵌入了一個 loop，這是因爲 CAS 是個代價比較高的操作，它需要獲得對應內存的獨佔訪問（exclusive access），我們希望失敗的時候只是簡單讀取 atomic 的狀態，只有符合條件的時候再去做獨佔訪問，進行 CAS。所以，看上去我們多做了一層循環，實際代碼的效率更高。

以下是兩個線程同步的過程，一開始 t1 拿到鎖，t2 spin，之後 t1 釋放鎖，t2 進入到臨界區執行：

通過上面的例子，相信你對 atomic 以及其背後的 CAS 有個初步的瞭解，如果你還想對 Rust 下使用 atomic 有更多更深入的瞭解，可以看 Jon Gjengset 最新一期 Crust of Rust: Atomics and Memory Ordering [6]。巧的是這周我計劃寫有關併發原語的文章，Jon 的視頻就出來了，幫我進一步夯實了關於 atomic 的知識。

上文中，爲了展示如何使用 atomic，我們製作了一個非常粗糙簡單的 SpinLock [7]。SpinLock，顧名思義，就是線程通過 CPU 空轉（spin，就像上文中的 while loop），來等待某個鎖可用的一種鎖。SpinLock 和 Mutex lock 最大的不同是，使用 SpinLock，線程在忙等（busy wait），而使用 Mutex lock，線程會在等待鎖的時候被調度出去，等鎖可用時再被被調度回來。

聽上去 SpinLock 似乎效率很低，但這要具體看鎖的臨界區的大小。如果臨界區要執行的代碼很少，那麼和 Mutex lock 帶來的上下文切換（context switch）相比，SpinLock 是值得的。在 Linux Kernel 中，很多時候，我們只能使用 SpinLock。

Rust 的 spin-rs crate [8] 提供了 spinlock 的實現。

那麼，atomic 除了做其它併發原語，還有什麼作用？

我個人用的最多的是做各種 lock-free 的數據結構。比如，我們需要一個全局的 id 生成器。我們當然可以使用 uuid 這樣的模塊來生成唯一的 id，但如果我們同時需要這個 id 是有序的，那麼 AtomicUsize 就是最好的選擇。你可以用 fetch_add 來增加這個 id，而 fetch_add 返回的結果就可以用於當前的 id。這樣，我們不需要加鎖，就得到了一個可以在多線程中安全使用的 id 生成器。

另外，atomic 還可以用於記錄系統的各種 metrics。比如我做的一個簡單的 in-memory Metrics 模塊：

use std::{
    collections::HashMap,
    sync::atomic::{AtomicUsize, Ordering},
};
// server statistics
pub struct Metrics(HashMap<&'static str, AtomicUsize>);
impl Metrics {
    pub fn new(names: &[&'static str]) -> Self {
        let mut metrics: HashMap<&'static str, AtomicUsize> = HashMap::new();
        for name in names.iter() {
            metrics.insert(name, AtomicUsize::new(0));
        }
        Self(metrics)
    }
    pub fn inc(&self, name: &'static str) {
        if let Some(m) = self.0.get(name) {
            m.fetch_add(1, Ordering::Relaxed);
        }
    }
    pub fn add(&self, name: &'static str, val: usize) {
        if let Some(m) = self.0.get(name) {
            m.fetch_add(val, Ordering::Relaxed);
        }
    }
    pub fn dec(&self, name: &'static str) {
        if let Some(m) = self.0.get(name) {
            m.fetch_sub(1, Ordering::Relaxed);
        }
    }
    pub fn snapshot(&self) -> Vec<(&'static str, usize)> {
        self.0
            .iter()
            .map(|(k, v)| (*k, v.load(Ordering::Relaxed)))
            .collect()
    }
}

它允許你初始化一個全局的 metrics 表，然後在程序的任何地方無鎖地操作相應的 metrics：

lazy_static! {
    pub(crate) static ref METRICS: Metrics = Metrics::new(&[
        "topics",
        "clients",
        "peers",
        "broadcasts",
        "servers",
        "states",
        "subscribers"
    ]);
}

Mutex

在併發處理中，一個核心的問題就是資源共享：軟件系統如何控制多個線程對同一個共享資源的訪問，使得每個線程可以在訪問共享資源的時候獨佔或者說互斥訪問（MUTual EXclusive access）？

我們知道，對於一個共享資源，如果所有線程只做讀操作，那麼無需互斥，大家隨時可以訪問，很多 immutable language（如 erlang/elixir）做了語言層面的只讀保證，確保了併發環境下的無鎖操作 [9]。這犧牲了一些效率（常見的 list/hashmap 需要使用 immutable data structure），額外做了不少內存拷貝，換來併發控制下的簡單輕靈。

然而一旦有任何一個或多個線程要修改共享資源，那麼不但寫者之間要互斥，讀寫之間也需要互斥。如果讀寫之間不互斥的話，那麼讀者輕則讀到髒數據，重則讀到已經被破壞的數據，導致 crash。比如讀者讀到鏈表裏的一個節點，而寫者恰巧把這個節點的內存釋放，如果不做互斥訪問，系統一定會崩潰。

用來解決這種讀寫互斥問題的一大基本工具就是 Mutex（RwLock 我們放下不表）。

上文中，我們製作的簡單的 spinlock，可以看做是一個廣義的 Mutex。然而，這種通過 spinlock 做互斥的實現方式有使用場景的限制：如果受保護的臨界區太大，那麼整體的性能會急劇下降， CPU 忙等，浪費資源還不幹實事，不適合作爲一種通用的處理方法。

更通用的解決方案是：當多個線程競爭同一個 Mutex 時，獲得鎖的線程得到臨界區的訪問，其它線程會被掛起，放入該 Mutex 上的一個等待隊列。當獲得鎖的線程完成工作，退出臨界區時，Mutex 會給等待隊列發一個信號，把隊列中第一個線程喚醒，於是這個線程可以進行後續的訪問。整個過程如下：

我們前面也講過，線程的上下文切換代價很大，所以頻繁將線程掛起再喚醒，會降低整個系統的效率。所以很多 Mutex 具體的實現會將 spinlock（確切地說是 spin wait）和線程掛起結合使用：線程的 lock 請求如果拿不到會先嚐試 spin 一會，然後再掛起添加到等待隊列。Rust 下的 parking_lot [10] 就是這樣實現的。

當然，這樣實現會帶來公平性的問題：如果新來的線程恰巧在 spin 過程中拿到了鎖，而當前等待隊列中還有其它線程在等待鎖，那麼等待的線程只能繼續等待下去，這不符合 FIFO，不適合那些需要嚴格按先來後到排隊的使用場景。爲此，parking_lot 提供了 fair mutex。

Mutex 的實現依賴於 CPU 提供的 atomic。你可以把 Mutex 想象成一個粒度更大的 atomic，只不過這個 atomic 無法由 CPU 保證，而是通過軟件算法來實現。

至於操作系統裏另一個重要的概念信號量（semaphore ），你可以認爲是 Mutex 更通用的表現形式。比如在新冠疫情下，圖書館要控制同時在館內的人數，如果滿了，其他人就必須排隊，出來一個才能再進一個。這裏，如果總人數限制爲 1，就是 Mutex，如果 > 1，就是 semaphore。大家可以想想可以怎麼實現 semaphore。也可以想想這樣的人數控制系統怎麼用信號量實現（提示：Rust 下 tokio 提供了 tokio::sync::Semaphore）。

Condvar

Mutex 解決了併發環境下共享資源如何安全訪問的問題，但它沒有解決一個更高層次的問題：如果這種訪問需要按照一定順序進行，該怎麼做？這個問題的典型的場景是生產者消費者模式：生產者生產出來內容後，需要有機制通知消費者可以消費。比如 socket 上有數據了，通知處理線程處理數據，處理完成之後，再通知 socket 收發的線程發送數據。更復雜的場景如我之前文章《透過 rust 探索系統的本原：併發篇》：數據接收，通知數據寫入，日誌寫滿，通知 S3 upload。在這個例子裏我使用了 channel 來完成，但 condvar 也適用：

在操作系統裏，condvar 是一種狀態：

等待（wait）：線程在隊列中等待，直到滿足某個條件
通知（notify）：當 condvar 的條件滿足時，當前線程通知其他等待的線程可以被喚醒。通知可是是單個通知，也可以是多個通知，甚至廣播（通知所有人）。

在實踐中，Condvar 往往和 Mutex 一起使用：Mutex 用於保證條件的讀寫時互斥的，Condvar 用於控制線程的等待和喚醒。

我們通過實現 mpsc channel 來看看 condvar 是如何使用的。如果你用過 channel，你知道 channel 創建後可以一端寫，一端讀，其內部共享一個 ring buffer（在 Rust 裏我們可以用 VecDequeue）。當 channel 裏沒有數據可讀時，讀者會掛起（block），而寫者寫入新的數據時，需要通知讀者恢復運行。這個過程，我們需要通過 Condvar 來實現。對於 mpsc channel 來說，channel 可以有多個寫者，一個讀者，這是最經典 channel。

首先我們定義讀者，寫者，以及 channel 函數本身：

pub struct Sender<T> {
    shared: Arc<Shared<T>>,
}
pub struct Receiver<T> {
    shared: Arc<Shared<T>>,
}
struct Inner<T> {
    queue: VecDeque<T>,
}
struct Shared<T> {
    inner: Mutex<Inner<T>>,
    available: Condvar,
    senders: AtomicUsize,
    receivers: AtomicUsize,
}
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let shared = Shared::default();
    let shared = Arc::new(shared);
    (
        Sender { shared: shared.clone() },
        Receiver { shared },
    )
}

這裏面，VecDeque 我們用 Mutex 保護，這是內部的 queue 的實現，讀者和寫者訪問 queue 需要互斥。available 是一個 Condvar，我們用來掛起或者喚醒讀者。

對於寫者，拿到鎖之後可以往 queue 裏面添加數據，添加完之後，要調用 notify_one 來喚醒可能掛起的讀者（如果沒有，notify_one 基本啥也不做）：

impl<T> Sender<T> {
    pub fn send(&mut self, t: T) -> Result<()> {
        if self.shared.receivers.load(Ordering::SeqCst) == 0 {
            return Err(anyhow!("no receiver left"));
        }
        let mut inner = self.shared.inner.lock().unwrap();
        inner.queue.push_back(t);
        drop(inner);
        self.shared.available.notify_one();
        Ok(())
    }
}

對於讀者，拿到鎖之後，可以從 queue 中取數據。這裏如果取不到數據，需要 wait 把自己掛起在 Condvar 上，等待寫者 notify_one ：

impl<T> Receiver<T> {
    pub fn recv(&mut self) -> Result<T> {
        let mut inner = self.shared.inner.lock().unwrap();
        loop {
            match inner.queue.pop_front() {
                Some(t) => return Ok(t),
                None if self.shared.senders.load(Ordering::SeqCst) == 0 => {
                    return Err(anyhow!("no sender left"))
                }
                None => {
                    inner = self.shared.available.wait(inner).unwrap();
                }
            }
        }
    }
}

注意在調用 wait 時，需要把當前拿住的鎖交給 Condvar，Condvar 會將其釋放，然後把讀者加入等待隊列，當有人 notify_one 時，隊列中的讀者會被喚醒，重新拿到鎖，然後繼續進行。

目前這個實現還有一個問題：如果寫者退出了，沒有人再寫數據，在隊列裏的讀者不會有人喚醒，所以我們還需要對 channel 所有的寫者做一個計數 —— 自然的，你會想到使用 atomic 來完成，這就是爲什麼 Shared<T> 裏有 senders 這樣一個 AtomicUsize 。有了這樣一個變量，我們可以在寫者被複制的時候增加 senders，在寫者退出（Drop）時，減少 senders。於是有：

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        let old = self.shared.senders.fetch_add(1, Ordering::AcqRel);
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}
impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        let old = self.shared.senders.fetch_sub(1, Ordering::AcqRel);
        if old <= 1 {
            self.shared.available.notify_one();
        }
    }
}

在所有寫者退出的時候，我們還需要喚醒在等待隊列中的讀者，所以我們要調用 notify_one 來做通知。注意 atomic fetch 得到的結果是改變前的值，所以這裏上一次是 1 的話，這次一減，所有寫者都沒了，所以我們要在這個時候通知。

剩下的，都是一些邊邊角角的工作，比如讀者退出後，寫者在往 queue 裏寫的時候需要返回錯誤等等。

以上是一個很簡單的 mpsc channel 的實現，裏面需要做性能優化的地方還很多。如果你對 channel 的實現感興趣，可以看 Rust 下 crossbeam channel [11] 或者 flume [12] 的實現。flume 沒有使用 Condvar 來做 signal，而是使用 thread::park 和 thread::notify 自己寫了一個簡單的 SyncSignal。

希望通過這個例子，你能對 Condvar 的使用有更深刻的認識。如果我們需要在線程間通過一定的條件來進行同步，那麼 Condvar 是一個不錯的選擇。此外，Condvar 還是一個非常不錯的，在多個線程間廣播的工具。

Channel

由於 golang 不遺餘力的推廣，channel 可能是最廣爲人知的併發手段。相對於 Mutex，channel 的抽象程度最高，接口最爲直觀，使用起來的心理負擔也沒那麼大。使用 Mutex 時，你需要很小心地避免死鎖，控制臨界區的大小，防止一切可能發生的意外。雖然在 Rust 裏，我們可以「無畏併發」（Fearless concurrency）—— 當我們的代碼編譯通過，那麼絕大多數併發問題都可以規避，但性能上的問題，邏輯上的死鎖還需要開發者照料。channel 把鎖封裝在了隊列寫入和讀取的小塊區域內，然後把讀者和寫者完全分離，使得讀者讀取數據和寫者寫入數據，對開發者而言，除了潛在的上下文切換外，完全和鎖無關，就像訪問一個本地隊列一樣。所以，對於大部分併發問題，我們都可以用 channel 或者類似的思想來處理（比如 actor model）。

channel 在具體實現的時候，根據不同的使用場景，會選擇不同的工具。上文中實現的一個簡單的 mpsc channel，我們使用了 Mutex + Condvar + VecDeque，那麼其它類型的 channel 呢？

我們大致捋一捋：

oneshot：這可能是最簡單的 channel，寫者就只發一次數據，而讀者也只讀一次。這種一次性的多個線程間的同步可以用 oneshot channel 完成。由於 oneshot 特殊的用途，實現的時候可以直接用 atomic swap 來完成。
rendezvous：很多時候，我們只需要通過 channel 來控制線程間的同步，並不需要發送數據。rendezvous channel 是 channel size 爲 0 的一種特殊情況。這種情況下，我們用 Mutex + Condvar 實現就足夠了。rendezvous channel 其實就是 Mutex + Condvar 的一個包裝，可以用 Condvar 代替。
bounded：bounded channel 有一個 queue，但 queue 有一個上限。一旦 queue 被寫滿了，寫者也需要被掛起等待。上文中我們實現的是一個 unbounded mpsc channel，寫者是可以無阻塞寫的，而 bounded channel 寫者有可能被阻塞。當阻塞發生後，讀者一旦讀取數據，需要 notify_one 通知寫者，喚醒某個寫者使其能夠繼續寫入。因此，我們可以用 Mutex + Condvar + VecDeque 來實現（大家感興趣的話，可以在上文例子的基礎上，將其修改爲 bounded channel）。如果不用 Condvar，可以直接使用 thread::park + thread::notify 來完成（flume 的做法），如果不用 VecDeque，也可以使用雙向鏈表或者其它的 ring buffer 的實現。
unbounded：queue 沒有上限，如果寫滿了，就自動擴容。Rust 的很多數據結構如 Vec ，VecDeque 都是自動擴容的。unbounded 和 bounded 相比，除了不阻塞寫者，其它實現都很類似。

所有這些 channel 類型，同步和異步的實現思路大同小異，主要的區別在於掛起 / 喚醒的對象。在同步的世界裏，掛起 / 喚醒的對象是線程；而異步的世界裏，是粒度很小的 task。

Mutex / Condvar 也是如此。

當我們做大部分複雜的系統設計時，channel 往往是最有力的武器，它除了可以讓數據穿梭於各個線程，各個異步任務間，其接口還可以很優雅地跟 stream 適配。如果說我們在做整個後端的系統架構時，着眼的是我們有哪些服務，服務和服務之間如何通訊，數據如何流動，服務和服務間如何同步；那麼在做某一個服務的架構時，着眼的是有哪些功能性的線程（異步任務），它們之間的接口是什麼樣子，數據是如何流動，如何同步。在這裏，channel 兼具接口，同步和數據流三種功能，所以我說是最有力的武器。

然而它不該是唯一的武器。我們面臨的真實世界的併發問題是多樣的，解決方案也應該是多樣的，計算機科學家們在過去的幾十年裏不斷探索，構建的一系列的併發原語，也說明了很難有一種銀彈解決所有問題。就連 Mutex 本身，在實現中，還會根據不同的場景做不同的妥協（比如做 faireness 的妥協），因爲這個世界就是這樣，魚與熊掌不可兼得，沒有完美的解決方案，只有妥協出來的解決方案。所以 channel 不是銀彈，actor model 不是銀彈，lock 不是銀彈。一門好的編程語言可以提供大部分場景下的最佳實踐（如 erlang/golang），但不該營造一種氣氛，只有某個最佳實踐纔是唯一方案。很不幸的是，golang 對 channel 的過分宣傳和癡迷使得很多程序員遇到問題就試圖用 channel 解決，頗有一種拿着錘子到處找釘子的感覺。

相反，Rust 提供幾乎你需要的所有解決方案，並且並不鼓吹他們的優劣，完全交由你按需選擇。我在用 Rust 撰寫多線程應用時，channel 是我的第一選擇，但我還是會在合適的時候使用 Mutex，RwLock，Semaphore，Condvar，Atomic 等工具，而不是試圖笨拙地用 channel 疊加 channel 來應對所有的場景。

賢者時刻

我們在學習某個知識時，我認爲最好的方式是拿破崙式的戰法：炮兵洗地，騎兵衝擊，最後由步兵掃尾和鞏固陣地。

比如對於 Mutex，我的炮兵是維基百科和有關 Mutex 的文獻（Linux 的 Futex 的介紹，LWN.net 相關的文檔，Rust std 裏關於 Mutex 的文檔等），通過這些內容，高屋建瓴地理解概念本身；騎兵是源代碼，比如 parking_lot（Rust std 關於 Mutex 的源碼實際包裝了 libc 的實現，所以直接看意義不太大）或者 libc 和 kernel 裏對應的實現，通過閱讀這些實現，你可以把理論和實際結合起來，並對業界的「最佳實踐」有一個不錯的理解；最後步兵是自己撰寫一個 high level 的實現，比如上圖，看着簡單，實現起來還是有很多細節需要處理 —— 尤其是讀代碼的時候你彷彿讀懂了，真要自己寫的時候，發現不是那麼一回事。當你這麼處理一輪之後，尤其層層遞進，用步兵鞏固陣地之後，這個知識就真真正正地成爲你自己的學問，成爲你的洞察力和判斷力的一部分。之後，你就可以回答幾乎任何與之相關的問題，即便這樣的問題你沒有學過，沒有經歷過，你也能找到一個分析框架，來解答這樣的問題。

參考資料

[1] Redlock: https://redis.io/topics/distlock

[2] data 使用 UnsafeCell<T>，然後在讀取的時候做 unsafe { &mut *self.data.get() }) 可以規避借用檢查，在多線程環境下獲得可變引用。

[3] CAS: https://en.wikipedia.org/wiki/Compare-and-swap

[4] Ordering: https://doc.rust-lang.org/std/sync/atomic/enum.Ordering.html

[5] std::memory_order: https://en.cppreference.com/w/cpp/atomic/memory_order

[6] Atomics and Memory Ordering: https://www.youtube.com/watch?v=rMGWeSjctlY

[7] spinlock: https://en.wikipedia.org/wiki/Spinlock

[8] spin-rs: https://github.com/mvdnes/spin-rs

[9] 語言的內核裏依然是需要修改內存的，所以依然需要對共享資源加鎖，但開發者層面不可見。

[10] parking lot：https://github.com/Amanieu/parking_lot

[11] Flume: https://github.com/zesterer/flume

[12] Crossbeam channel：https://docs.rs/crossbeam-channel

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/fJO-rCgL9N5fPvrqtHe9Ug