Skip to main content
开始使用 tokio

1 tokio 异步编程模型

•1 min•195 words
首先学好rust的异步部分

关于 future trait,await/async 关键字的作用。

异步运行时原理

首先我们需要一个线程用于运行异步运行时。通常这个线程就是程序启动的主线程。然后将其他代码全部包装为异步任务。运行时会拥有线程池来执行这些任务。

tokio 启动

使用宏来启动运行时

[!summary] 使用#[tokio::main] 宏来启动运行时
这是一个属性宏,能将所标注的 main 函数重新包裹。会被包裹成类似下面的代码,用以启动异步运行时。

#[tokio::main]
async fn main() {
    // 异步代码
}
// 展开后应该是
fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            // 原异步 main 的内容
        })
}
另外,Tokio 具有很多功能,并非所有应用程序都需要所有功能

应用程序可以决定只选择它使用的功能。在添加tokio依赖的时候选择需要的feature。

tokio 生成任务

网络操作是 io 异步操作收益最大的部分之一

tokio 提供了异步化的网络接口,可以让网络操作在tokio中以异步任务的方式运行。
这些API都在tokio::net模块下

无逻辑顺序关系的任务,适合并发执行

在异步任务中,或者任何任务中,我们只能并发执行那些没有逻辑顺序关系的任务。因为并发执行不保证完成的顺序性,也不保证执行过程中的顺序。

每个任务都是一个独立的类型,这些类型必须是’static的

'static 意味着无论在什么时候执行这个任务,其所类型都能从运行时内存结构中的静态数据区域找到。不会出现发现类型不完整的情况。
.
通常是因为任务的future结构体字段不’static,进一步是因为 异步代码块或者函数使用了外部引用,无法保证引用的生命周期,最好move进来之后,让数据成为该future结构体的’static数据。

tokio::spawn生成新任务

传递给这个函数一个future结构体实例,然后就会将一个异步任务加入异步任务队列。
.
由tokio::spawn 生成的任务必须实现 Send。这允许 Tokio 运行时在任务在 .await 处挂起时在线程之间移动任务。也就是多线程工作窃取需要。
.
应该多多生成异步任务,因为开销非常的小。任务队列只需要一个future对象分配和 64 字节的内存指针。应用程序应该随意生成数千个,甚至数百万个任务。

Send与Sync约束

在《the book》中无畏并发这一节,在最后会有关于Send与Sync trait的内容。Send保证类型值的所有权可以在线程间传送,因为有的类型比如rc,裸指针,LocalSet这些移动所有权之后无法保证功能正常。
.
Sync保证类型的引用在多线程访问与修改的时候不会出现数据竞争。至于生命周期的问题,由编译器检查以及通过Scope约束线程生命周期。

什么时候没有自动实现Send

在异步任务中,之所以需要Send是应为要在多线程之间移动。而移动的时机只会出现在.await交出执行器控制权的时候。这个时间点可能发生线程移动。所以在这个时间点还保持存在的变量其类型需要实现Send

共享状态

在不同的线程中共享状态

需要解决的问题就是数据竞争,通常有两种重量散与轻量CAS。CAS虽然轻量,但是无法占据。如果业务需要占据数据来保证正确性,那么必须使用锁。

两种常用的共享状态选择

1 使用Mutex锁来避免数据竞争。
2 生成一个任务来管理状态,并使用消息传递来操作它。(因为消息传递本身具有同步操作)。
.
对于简单关系状态数据,更推荐使用第一种方法,简单。但是注意锁的时间存在短才是简单的意义。
.
对于需要异步工作(如 I/O 原语)的状态,复杂的状态关系,则使用第二种方法。这样对状态操作的时候不会影响其他任务,并且只有状态任务能操作状态,避免了数据竞争。

同步锁与异步锁

同步锁指std提供的mutex,基于操作系统提供的锁原语,当线程无法获得时阻塞线程。
异步锁指tokio提供的mutex,阻塞的不是线程而是任务,在异步锁释放之后,按顺序唤醒阻塞任务,不会耽误执行器线程。
.
异步锁的主要好处是可以在异步锁持有的时候.await,这样执行器依旧能执行异步任务,不过还是会有循环等待死锁的情况。
.
异步锁的性能弱于同步锁,所以在没有好处的情况下不要使用异步锁。

跨.await的状态需要Send,并且无法显式drop !Send状态

比如锁,通常同步锁如果.await会导致线程直接无法运行。但是因为std提供的mutex不是Send trait类型,那么其实根本无法通过编译。(但是某些锁实现了Send需要小心)
.
另外的一个问题是显式drop在编译器判断当前Future的Send不管用,因为编译器当前仅根据作用域信息计算 future 是否为Send。编译器有望在未来更新以支持显式删除它,但目前,您必须显式使用作用域。

锁的最佳使用

确保不跨.await:在结构体中,在同步方法中使用锁

避免了在操作锁的时候插入.await,避免外界访问到锁,控制锁的使用方法。

use std::sync::Mutex;

struct CanIncrement {
    mutex: Mutex<i32>,
}
impl CanIncrement {
    // This function is not marked async.
    fn increment(&self) {
        let mut lock = self.mutex.lock().unwrap();
        *lock += 1;
    }
}

async fn increment_and_do_stuff(can_incr: &CanIncrement) {
    can_incr.increment();
    do_something_async().await;
}
current_thread 运行时风格

一个轻量级的单线程运行时。单线程使得锁永远不会被争用,但是这种运行时只对少量任务时有好效果。

互斥锁分片:更小的锁定粒度

通常我们会锁定整个对象,就像Rust的切片一样,不允许同时访问实际上根本不交叉的区域。
实际上我们可以减小锁的粒度,更精准的锁住必要的数据,让不必要的数据能自由。

当然,给每一个基本数据类型添加锁是不现实的,这会大大冗余拖慢效率。我们可以分片,将合适数量的数据分配一个锁。就像国家分省,省内分市一样。

let mut ShardedDb = Arc<Vec<Mutex<HashMap<String, Vec<u8>>>>>;

分片技术的关键是如何将数据均匀(指访问频率)分片,以及稳定扩容。

锁分片的生产级实现

dashmap crate 提供了更复杂的分片哈希映射的实现。您可能还想查看诸如 leapfrog 和 flurry 之类的并发哈希表实现,后者是 Java 的 ConcurrentHashMap 数据结构的端口。

异步任务管理状态

任务管理状态:通过消息传递间接操作状态的好处

1 限制了胡乱访问,避免了锁的长期持有
2 通道充当缓冲区。
3 让状态更容易拓展

Tokio 提供的信道

1mpsc:多生产者、单消费者通道。可以发送多个值。
2 oneshot:单生产者、单消费者通道。可以发送单个值。
3 broadcast:多生产者、多消费者。可以发送多个值。每个接收者都会看到每个值。
4 watch:多生产者、多消费者。可以发送多个值,但不保留历史记录。接收者只会看到最近的值。

另外还有一些信道,一个多生产者多消费者通道,其中每个消息只有一个消费者看到,你可以使用 async-channel crate。还有一些通道用于异步 Rust 之外,例如 std::sync::mpsc 和 crossbeam::channel。这些通道通过阻塞线程来等待消息,这在异步代码中是不允许的。

tokio::mpsc::channel的使用

tokio 要求channel的rx独占,因此不通过rx的clone以及使用需要&mut ,所以我们最好使用(sx,mut rx) 来接收。
.
channel函数需要指明 buffer size,接收usize参数作为buffer size当通道中积压的信息超过这个数,send(xxx).await方法将陷入异步睡眠直到通道有多余容量。

Faq

为什么tokio的rec返回Option,以None为通道关闭?

Receiver的rec方法返回的直接就是Option,当返回None的时候表示通道无内容,所以Sender关闭。因为当Sender不关闭且通道无内容的时候异步rec方法会在中途异步阻塞不会返回内容。因为在异步中,以Steam约定,Some(item) 表示还有元素,None 表示流结束。这与 recv 返回 Option<T> 完全一致。

这与std提供的同步rec方法返回值不同,Result<T,E>使用Err表示通道关闭而非None。std:设计于 Rust 1.0 之前,如果 std 的 mpsc 是今天重新设计的,说不定也会用 Option。但 std 要保证向后兼容,改不了。

使用通道来返回结果

[!summary] 使用oneshot通道来返回结果
因为使用mpsc的通道,是单向的,我们需要从状态管理任务中获得状态。而oneshot通道是最合适的,单向单容量。
.
只需要在mpsc中发送消息中携带oneshot的Sender即可。

oneshot 通道的特点

容量为1,send 将始终立即失败或成功,而无需任何形式的等待。当接收者一半被丢弃时,在 oneshot 通道上发送值会返回 Err。这表示接收者不再对响应感兴趣。