Module sync

展开描述

用于异步上下文的同步原语。

Tokio 程序往往被组织为一组 tasks，其中每个任务独立运行，可能在不同的物理线程上执行。本模块提供的同步原语允许这些独立的任务相互通信。

§Message passing

Tokio 程序中最常见的同步形式是消息传递。两个任务独立运行并相互发送消息以进行同步。这样做的好处是避免了共享状态。

消息传递是使用通道实现的。通道支持从一个生产者任务向一个或多个消费者任务发送消息。 Tokio 提供了几种不同风格的通道。每种通道风格支持不同的消息传递模式。当通道支持多个生产者时，许多独立的任务可以发送消息。当通道支持多个消费者时，许多不同的独立任务可以接收消息。

Tokio 提供了许多不同的通道风格，因为不同的消息传递模式最好由不同的实现来处理。

§`oneshot` channel

oneshot 通道支持从一个单个生产者向单个消费者发送单个值。此通道通常用于将计算结果发送给等待者。

示例：使用 oneshot 通道接收一个计算的结果。

use tokio::sync::oneshot;

async fn some_computation() -> String {
    "represents the result of the computation".to_string()
}

let (tx, rx) = oneshot::channel();

tokio::spawn(async move {
    let res = some_computation().await;
    tx.send(res).unwrap();
});

// Do other work while the computation is happening in the background

// Wait for the computation result
let res = rx.await.unwrap();

注意，如果任务在终止前将计算结果作为其最终动作产生，则可以使用 JoinHandle 来接收该值，而不是为 oneshot 通道分配资源。在 JoinHandle 上等待会返回 Result。如果任务 panic，则 Joinhandle 会产生 Err，其中包含 panic 的原因。

示例：

async fn some_computation() -> String {
    "the result of the computation".to_string()
}

let join_handle = tokio::spawn(async move {
    some_computation().await
});

// Do other work while the computation is happening in the background

// Wait for the computation result
let res = join_handle.await.unwrap();

§`mpsc` channel

mpsc 通道支持从多个生产者向单个消费者发送多个值。此通道通常用于向任务发送工作或接收多个计算的结果。

如果你想从单个生产者向单个消费者发送多条消息，也应该使用此通道。没有专用的 spsc 通道。

示例：使用 mpsc 增量流式传输一系列计算的结果。

use tokio::sync::mpsc;

async fn some_computation(input: u32) -> String {
    format!("the result of computation {}", input)
}

let (tx, mut rx) = mpsc::channel(100);

tokio::spawn(async move {
    for i in 0..10 {
        let res = some_computation(i).await;
        tx.send(res).await.unwrap();
    }
});

while let Some(res) = rx.recv().await {
    println!("got = {}", res);
}

mpsc::channel 的参数是通道容量。这是任何给定时间通道中可存储的待接收值的最大数量。正确设置此值对于实现健壮的程序至关重要，因为通道容量在处理背压方面起着关键作用。

一种常见的资源管理并发模式是派生一个专门管理该资源的任务，并使用其他任务之间的消息传递与该资源进行交互。该资源可以是任何可能无法并发使用的东西。一些示例包括 socket 和程序状态。例如，如果多个任务需要通过单个 socket 发送数据，则派生一个任务来管理该 socket，并使用通道进行同步。

示例：通过消息传递从多个任务通过单个 socket 发送数据。

use tokio::io::{self, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut socket = TcpStream::connect("www.example.com:1234").await?;
    let (tx, mut rx) = mpsc::channel(100);

    for _ in 0..10 {
        // Each task needs its own `tx` handle. This is done by cloning the
        // original handle.
        let tx = tx.clone();

        tokio::spawn(async move {
            tx.send(&b"data to write"[..]).await.unwrap();
        });
    }

    // The `rx` half of the channel returns `None` once **all** `tx` clones
    // drop. To ensure `None` is returned, drop the handle owned by the
    // current task. If this `tx` handle is not dropped, there will always
    // be a single outstanding `tx` handle.
    drop(tx);

    while let Some(res) = rx.recv().await {
        socket.write_all(res).await?;
    }

    Ok(())
}

mpsc 和 oneshot 通道可以组合使用，以提供与共享资源的请求/响应类型同步模式。派生一个任务来同步资源，并等待通过 mpsc 通道接收到的命令。每个命令都包含一个 oneshot Sender，命令的结果通过它发送。

示例：使用任务同步一个 u64 计数器。每个任务发送一个“取值并自增”命令。递增前的计数器值通过提供的 oneshot 通道发送。

use tokio::sync::{oneshot, mpsc};
use Command::Increment;

enum Command {
    Increment,
    // Other commands can be added here
}

let (cmd_tx, mut cmd_rx) = mpsc::channel::<(Command, oneshot::Sender<u64>)>(100);

// Spawn a task to manage the counter
tokio::spawn(async move {
    let mut counter: u64 = 0;

    while let Some((cmd, response)) = cmd_rx.recv().await {
        match cmd {
            Increment => {
                let prev = counter;
                counter += 1;
                response.send(prev).unwrap();
            }
        }
    }
});

let mut join_handles = vec![];

// Spawn tasks that will send the increment command.
for _ in 0..10 {
    let cmd_tx = cmd_tx.clone();

    join_handles.push(tokio::spawn(async move {
        let (resp_tx, resp_rx) = oneshot::channel();

        cmd_tx.send((Increment, resp_tx)).await.ok().unwrap();
        let res = resp_rx.await.unwrap();

        println!("previous value = {}", res);
    }));
}

// Wait for all tasks to complete
for join_handle in join_handles.drain(..) {
    join_handle.await.unwrap();
}

§`broadcast` channel

broadcast 通道支持从多个生产者向多个消费者发送多个值。每个消费者将接收每个值。此通道可用于实现 pub/sub 或“聊天”系统中常见的 “扇出”模式。

此通道的使用频率往往低于 oneshot 和 mpsc，但仍然有其用例。

如果你想从单个生产者向多个消费者 broadcast 值，也应该使用此通道。没有专用的 spmc broadcast 通道。

基本用法

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    assert_eq!(rx1.recv().await.unwrap(), 10);
    assert_eq!(rx1.recv().await.unwrap(), 20);
});

tokio::spawn(async move {
    assert_eq!(rx2.recv().await.unwrap(), 10);
    assert_eq!(rx2.recv().await.unwrap(), 20);
});

tx.send(10).unwrap();
tx.send(20).unwrap();

§`watch` channel

watch 通道支持从多个生产者向多个消费者发送多个值。但是，仅最近的值存储在通道中。当发送新值时，消费者会收到通知，但不能保证消费者将看到所有值。

watch 通道类似于 broadcast 通道，但 capacity 为 1。

watch 通道的用例包括广播配置更改或发出程序状态更改的信号，例如过渡到关闭。

示例：使用 watch 通道通知任务配置更改。在本例中，配置文件被定期检查。当文件更改时，配置更改会向消费者发出信号。

use tokio::sync::watch;
use tokio::time::{self, Duration, Instant};

use std::io;

#[derive(Debug, Clone, Eq, PartialEq)]
struct Config {
    timeout: Duration,
}

impl Config {
    async fn load_from_file() -> io::Result<Config> {
        // file loading and deserialization logic here
    }
}

async fn my_async_operation() {
    // Do something here
}

// Load initial configuration value
let mut config = Config::load_from_file().await.unwrap();

// Create the watch channel, initialized with the loaded configuration
let (tx, rx) = watch::channel(config.clone());

// Spawn a task to monitor the file.
tokio::spawn(async move {
    loop {
        // Wait 10 seconds between checks
        time::sleep(Duration::from_secs(10)).await;

        // Load the configuration file
        let new_config = Config::load_from_file().await.unwrap();

        // If the configuration changed, send the new config value
        // on the watch channel.
        if new_config != config {
            tx.send(new_config.clone()).unwrap();
            config = new_config;
        }
    }
});

let mut handles = vec![];

// Spawn tasks that runs the async operation for at most `timeout`. If
// the timeout elapses, restart the operation.
//
// The task simultaneously watches the `Config` for changes. When the
// timeout duration changes, the timeout is updated without restarting
// the in-flight operation.
for _ in 0..5 {
    // Clone a config watch handle for use in this task
    let mut rx = rx.clone();

    let handle = tokio::spawn(async move {
        // Start the initial operation and pin the future to the stack.
        // Pinning to the stack is required to resume the operation
        // across multiple calls to `select!`
        let op = my_async_operation();
        tokio::pin!(op);

        // Get the initial config value
        let mut conf = rx.borrow().clone();

        let mut op_start = Instant::now();
        let sleep = time::sleep_until(op_start + conf.timeout);
        tokio::pin!(sleep);

        loop {
            tokio::select! {
                _ = &mut sleep => {
                    // The operation elapsed. Restart it
                    op.set(my_async_operation());

                    // Track the new start time
                    op_start = Instant::now();

                    // Restart the timeout
                    sleep.set(time::sleep_until(op_start + conf.timeout));
                }
                _ = rx.changed() => {
                    conf = rx.borrow_and_update().clone();

                    // The configuration has been updated. Update the
                    // `sleep` using the new `timeout` value.
                    sleep.as_mut().reset(op_start + conf.timeout);
                }
                _ = &mut op => {
                    // The operation completed!
                    return
                }
            }
        }
    });

    handles.push(handle);
}

for handle in handles.drain(..) {
    handle.await.unwrap();
}

§State synchronization

其余的同步原语专注于同步状态。这些是 std 提供的版本的异步等效物。它们的工作方式与 std 对应物类似，但会异步等待，而不是阻塞线程。

Barrier 确保多个任务将相互等待以到达程序中的某个点，然后再一起继续执行。
Mutex 互斥机制，确保一次最多一个线程能够访问某些数据。
Notify 基本任务通知。 Notify 支持在不发送数据的情况下通知接收任务。在这种情况下，任务唤醒并继续处理。
RwLock 提供了一种互斥机制，允许同时有多个读者，但一次只允许一个写者。在某些情况下，这比 mutex 更高效。
Semaphore 限制并发量。 semaphore 持有一定数量的 permit，任务可以请求它们以进入临界区。 Semaphore 可用于实现任何类型的限制或限定。

§Runtime compatibility

本模块提供的所有同步原语都是与运行时无关的。你可以自由地将它们在 Tokio 运行时的不同实例之间移动，甚至可以从非 Tokio 运行时使用它们。

在 Tokio 运行时中使用时，同步原语参与协作式调度以避免饥饿。当从非 Tokio 运行时使用时，此功能不适用。

作为例外，以 _timeout 结尾的方法不是与运行时无关的，因为它们需要访问 Tokio 计时器。有关其用法的更多信息，请参阅每个 *_timeout 方法的文档。

模块§

broadcast: 一个多生产者、多消费者的广播队列。每个发送的值都会被所有消费者看到。
futures: 命名的 future 类型。
mpsc: 用于在异步任务之间发送值的多生产者、单消费者队列。
oneshot: oneshot channel 用于在异步任务之间发送单个消息。可使用 channel 函数创建 Sender 和 Receiver 句柄对，二者构成该 channel。
watch: 一个多生产者、多消费者的 channel，仅保留最近一次发送的值。

结构体§

AcquireError: 由 Semaphore::acquire 函数返回的错误。
Barrier: Barrier（屏障）使多个 task 能够同步开始某段计算。
BarrierWaitResult: 当 Barrier 中所有 task 都在 wait 处汇合时，wait 返回一个 BarrierWaitResult。
MappedMutexGuard: 对持有的 Mutex 通过 MutexGuard::map 应用函数后得到的句柄。
Mutex: 一个异步的、类似 Mutex 的类型。
MutexGuard: 持有的 Mutex 的句柄。由于实现了 Send，可以在任意 .await 点持有该 guard。
Notify: 通知单个 task 唤醒。
OnceCell: 一个线程安全的 cell，只能写入一次。
OwnedMappedMutexGuard: 对持有的 Mutex 通过 OwnedMutexGuard::map 应用函数后得到的 owned 句柄。
OwnedMutexGuard: 持有的 Mutex 的 owned 句柄。
OwnedRwLockMappedWriteGuard: Owned RAII 结构，在 drop 时释放锁的独占写访问。
OwnedRwLockReadGuard: Owned RAII 结构，在 drop 时释放锁的共享读访问。
OwnedRwLockWriteGuard: Owned RAII 结构，在 drop 时释放锁的独占写访问。
OwnedSemaphorePermit: 来自 semaphore 的 owned permit。
RwLock: 一个异步的读者-写者锁。
RwLockMappedWriteGuard: RAII 结构，在 drop 时释放锁的独占写访问。
RwLockReadGuard: RAII 结构，在 drop 时释放锁的共享读访问。
RwLockWriteGuard: RAII 结构，在 drop 时释放锁的独占写访问。
Semaphore: 执行异步 permit 获取的计数信号量（counting semaphore）。
SemaphorePermit: 来自 semaphore 的 permit。
SetOnce: 一个线程安全的 cell，只能写入一次。
SetOnceError: 可由 SetOnce::set 返回的错误。
TryLockError: 由 Mutex::try_lock、RwLock::try_read 和 RwLock::try_write 函数返回的错误。

枚举§

SetError: 可由 OnceCell::set 返回的错误。
TryAcquireError: 由 Semaphore::try_acquire 函数返回的错误。

Module sync

Module sync 复制项目路径

§Message passing

§oneshot channel

§mpsc channel

§broadcast channel

§watch channel

§State synchronization

§Runtime compatibility

模块§

结构体§

枚举§

Module sync

§`oneshot` channel

§`mpsc` channel

§`broadcast` channel

§`watch` channel