Rust并发控制之Semaphore-两线程交替打印

newbmiao

发布于 2023-11-27 04:33:41

49800

代码可运行

文章被收录于专栏：学点Rust学点Rust

运行总次数：0

代码可运行

信号量（Semaphore）是一种对资源并发访问控制的方式。

区别于互斥锁（Mutex）是对共享资源的独占访问，Semaphore 允许指定多个并发访问共享资源。

就是说 Semaphore 像一个持有令牌（permit/token）的桶，每一个并发访问需要持有（acquire）一个令牌来访问共享资源，

当没有令牌时，没法访问共享资源，直到有新的令牌加入（add）或者原来发出的令牌放回（release）桶中。

接下来，我们尝试用通过用它来实现两个线程交替打印 1 和 2，来更直观了解如何使用 semaphore

Rust std 库中没有正式发布的 semaphore（std::sync::Semaphore 在 1.7.0 废弃了）。下边用 tokio 库提供的 semaphore

首先安装 tokio 库

# 手动添加tokio到cargo.toml
# 或使用cargo-add: cargo add tokio --features sync,macros,rt-multi-thread
[dependencies]
tokio = { version = "1.34.0", features = ["sync", "macros", "rt-multi-thread"] }

先来一版常规实现，初始化一个只有一个令牌的 semahore，两个线程去并发持有令牌，用后释放（通过 drop）令牌，实现交替打印

use std::sync::Arc;
use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(1));
    let cnt = 3;
    let semaphore2 = semaphore.clone();

    let t1 = tokio::spawn(async move {
        for _ in 0..cnt {
            let permit = semaphore.acquire().await.unwrap();
            print!("1 ");
            // 可不写，离开scope时自动释放，放回令牌桶
            drop(permit);
        }
    });

    let t2 = tokio::spawn(async move {
        for _ in 0..cnt {
            // 或用 _ ignore返回值，即时回收令牌
            let _ = semaphore2.acquire().await.unwrap();
            print!("2 ");
        }
    });

    tokio::try_join!(t1, t2).unwrap();
}

乍看没什么问题，但是打印其实不一定是1 2 1 2 1 2的顺序。

原因很简单，我们只是约束了令牌同时只能有一个线程获取到，但是没有约束谁先谁后啊。所以其实没有实现交替打印。

怎么交替打印呢？

要控制顺序，我们可以让每个线程所持有的 semaphore 里的令牌时动态增加和消耗，然后一个令牌桶数量的增加滞后于另一个。

增加可以用 add_permits，消耗后不放回可以用 forgot，代码如下：

use std::sync::Arc;
use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    // 线程1的令牌桶1初始一个令牌，可以先打印1
    let semaphore = Arc::new(Semaphore::new(1));
    let cnt = 3;
    let semaphore2 = semaphore.clone();

    // 线程2的令牌桶2初始没有令牌，直到1打印后增加令牌
    let semaphore_wait = Arc::new(Semaphore::new(0));
    let semaphore_wait2 = semaphore_wait.clone();

    let t1 = tokio::spawn(async move {
        for _ in 0..cnt {
            let permit = semaphore.acquire().await.unwrap();
            print!("1 ");
            // 消耗令牌，不放回令牌桶1
            permit.forget();
            // 令牌桶2增加令牌，可以打印2
            semaphore_wait2.add_permits(1);
        }
    });

    let t2 = tokio::spawn(async move {
        for _ in 0..cnt {
            let permit = semaphore_wait.acquire().await.unwrap();
            print!("2 ");
            // 消耗令牌，不放回令牌桶2
            permit.forget();
            // 令牌桶1增加令牌，可以打印1
            semaphore2.add_permits(1);
        }
    });

    tokio::try_join!(t1, t2).unwrap();
}