Rust中的异步

发布时间：2023-03-03 17:09
最后更新：2024-06-20 22:38
所属分类：
Rust

自从Rust 1.36在标准库中引入了std::future::Future特征，异步就在Rust应用的各个领域遍地开花。

最先让我们接触到异步编程的，就是Tokio这个框架。基于现在版本的Rust标准库中的定义，已经有Tokio和async-std两个常用的异步框架供我们使用了，但是无论选择哪个框架，其核心都是基于标准库中提供的std::future::Future特征。

但是在std::future模块中，Rust还提供了一些用于快速实现异步函数的内容。

异步是当前许多编程语言中对于并发的一种新的实现形式。与传统的多线程编程不同，异步往往允许在一个线程上进行调度，从而使用更少的资源实现性能更高的任务调度，而且相比多线程编程，异步可以更好的利用现代多核心CPU。在Rust语言的初期是不支持异步的，然而在Tokio等异步库的推动下，Rust终于在标准库中提供了一套标准特征用来描述和供第三方库实现，但Rust标准库中并没有提供一套完整的异步实现。

Future特征

Future特征的目的就是用来描述一个在未来一定可以获取到的内容。类似于其他语言中的promise、delay、deferred等，Future就是对未来结果的一个包装代理。

在Rust中，Future特征其实并不复杂，首先来看一下它在标准库中的定义。

1
2
3
4
5

pub trait Future {
  type Output;

  fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

在这个特征定义中出现了不少内容，现在来逐一研究一下。

Poll枚举

Poll枚举代表的是Future当前的运行状态，其中提供的状态有两个。

  • Poll::Ready(T)，用于表示Future已经执行结束，将返回一个T类型的返回值。
  • Poll::Pending，用于表示Future当前还没有完成执行。如果Future选择返回Poll::Pending那么就需要确保自己被列入了唤醒队列，以使自己可以被再次唤醒。

Context结构与Waker结构

Context位于标准库中的std::task模块中，这个结构从它的名称上就可以看出来，它提供的是一个异步任务的运行环境上下文。Context在目前只是提供了对于&Waker结构类型的包装，可以允许便捷的访问公共的&Waker。Context主要提供了以下两个方法供使用。

  • fn from_waker(waker: &'a Waker) -> Context<'a>，这个方法会创建一个Context实例，并将一个&Waker引用包装起来。这个方法在使用的时候需要注意，被包装的&Waker引用，其生命期需要至少长于Context实例的生命期。
  • fn waker(&self) -> &'a Waker，这个方法会返回Context中包装的&Waker引用。

在Context中包装的Waker结构类型的实例其实并不难理解它的用途。它的主要功能就是通知执行器唤醒一个被挂起的任务。Waker实例一般都是由执行器创建并包装在Context中的。一个Future实例在执行的时候，如果选择返回Poll::Pending进入等待，调用Waker.wake()方法表示Future已经产生了一些进展，需要执行器关注并处理。

Pin特征

Pin特征在之前的文章中已经提到过了，这个特征的主要功能就是将被包装的对象固定在内存中，使其不可被移动。那在Future的poll方法中为什么要使用Pin来修饰&mut self的引用呢？

首先要了解的一个情况就是有一种结构叫“自引用结构”，这种结构在其他的语言中并不少见，但是这个结构在Rust中，却不是安全的。例如有以下一个结构。

1
2
3
4

struct Book {
  name: String,
  name_ref: &String,
}

在这个结构中，name_ref字段引用了name字段，这样在Book结构初始化的时候，一切都是正常的，name_ref字段中保存的也是name字段的地址。但是Rust会在所有权转移的时候在内存中移动值，所以当发生所有权转移的时候，name_ref中保存的地址值，就不再是name字段真正的地址值了，换句话说，谁也不知道此时的name_ref究竟引用的是哪里的值。在这种情况下实际就引发了比较重大的安全隐患。为了解决这种自引用类型带来的安全隐患，Rust引入了Pin、Unpin这一系列特征。

Unpin特征是非常好理解的，基本上所有Rust中的基本类型都是属于这一类的，它们在内存中的移动是安全的。

Unpin是一个自动特征（auto trait），这表示你不需要担心它是怎么实现的，在哪里实现的。你所需要做的事情就是在不确定这个类型是否可以在内存中安全移动的时候，查阅文档看它是否实现了Unpin特征。

Pin则是用来包装一个指针，并且会阻止这个指针的移动。一个类型被Pin包装以后，你就不能通过Pin<T>获取被包装类型的所有权，所以也就无法移动被包装的类型T。换一种更容易理解的说法就是Pin是一个持有被包装类型T的代理，你如果想访问T，那么就必须通过代理来完成，而你虽然拥有代理的所有权，但是并不能取得被代理内容的所有权。

在绝大部分使用Future特征的场景中，实现了Future特征的异步运行结果实例基本上都是以匿名实例出现的，这就会让匿名Future实例在内存中被移动，而且我们又不可能保证其中绝对不会存在自引用。这时就是Pin发挥其作用的时候了。Pin可以保证匿名Future实例在移动过程中，其中的所有自引用都不会出现问题。

使用Pin包装的内容是不能直接访问的，在访问的时候可以通过Pin::new()再包裹一层或者使用.as_def()来获取其中的内容。

其他可用特征与结构

以Future特征为基础，std::future模块还提供了其他若干比较实用的数据结构和功能。

IntoFuture特征

IntoFuture特征所表示的功能就跟其字面意思一样，表示实现了这个特征的数据结构可以被转换为一个Future。这个特征常常用在.await调用上，.await会首先调用IntoFuture中的.into_future()方法将实例转换未一个Future，然后再将其列入到异步处理中。

PollFn结构

PollFn结构跟FnOnce、Fn等结构一样，是用来描述函数类型的，只是PollFn描述的是一个会返回Poll结构类型的函数。std::future里提供了一个poll_fn()函数，可以用来快速创建一个PollFn类型的函数。PollFn实例可以直接使用.await进行异步执行。

poll_fn()函数的签名如下：

1
2
3

pub fn poll_fn<T, F>(f: F) -> PollFn<F>
where
  F: FnMut(&mut Context<'_>) -> Poll<T>,

从poll_fn()的签名可以看出来，这个函数可以将一个FnMut(&mut Context<'_>) -> Poll<T>类型的胡转换为一个PollFn的实现了Future特征的结构。

Pending结构

Pending结构是一个特殊的实现了Future的结构，它表示Future将永远不会结束。跟PollFn结构一样，Pending结构可以实用std::future中提供的pending()函数来创建。

Ready结构

Ready结构的功能与Pending结构的功能正好相反，Ready结构将创建一个立刻结束的实现Future的结构，同样的，它可以实用std::future中提供的ready()函数来创建。

编写异步程序

异步程序的编写有一个特点，如果程序中的一个函数或者语句块返回了Future，那么所有调用这个函数或者语句块的位置也都将返回Future。所以如果要编写实用了异步的程序，那么推荐整个程序都采用异步结构。

使用异步运行时

任何一个异步程序都需要一个异步运行时的支持，目前Rust中最为常用的异步运行时主要有Tokio和async-std两个。其中Tokio作为一款发展了很久的异步运行时，生态环境是非常好的，再实用过程中有大量的功能库可供使用。相比之下，async-std的特点则是足够新，完全基于std::future模块实现，没有什么历史包袱，但缺点也是太新，生态环境有些不足，对于各个功能库的选择余地小。

异步运行时的启动都是采用宏标记在main()函数上的，两个运行时在启动运行时的时候，书写方法基本上是相同的。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// Tokio运行时的启动
#[tokio::main]
async fn main() {
  // 书写异步处理过程
}

// async-std运行时的启动
#[async_std::main]
async fn main() {
  // 书写异步处理过程
}

从上面这两个示例中可以看出，异步运行时的启动就是依靠#[tokio::main]这样的宏标记，而且不同的运行时基本上都提供了相同的宏。

编写异步函数

其实编写一个异步函数并不需要使用到Future特征，函数只需要使用async关键字进行标记就可以了。例如编写一个访问Redis数据库中的指定键的函数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

async fn get_value(key: &str) -> Result<String> {
  let client = redis::Client::open("redis://127.0.0.1/");
  let mut connection = client.get_async_connection().await?;

  let result = redis::cmd("GET")
    .arg(&["key1"])
    .await?
    .to_string();

  Ok(result)
}

async关键字与fn关键字结合的时候，其用途是定义一个异步函数。async关键字还可以直接声明一个语句块，当async修饰一个语句块的时候，会自动将语句块的返回结果转换为Future<Output = T>。例如：

1
2
3
4
5
6

fn get_value() -> impl Future<Output = usize> {
  async {
    let x: usize = get_some_random_value().await;
    x
  }
}

在编写异步函数或者异步代码块的时候，需要牢记的是async标记的异步代码块是懒惰的，直到被执行器poll或者使用.await调用才会执行，并且使用.await调用时，如果Future没有完成，那么.await就会让出当前线程的控制权，当Future调用了Waker.wake()的时候，执行器才会重新继续运行Future，如此循环直到Future完成运行。

手动启动异步任务

除了可以使用.await启动异步任务以外，异步任务还可以直接在运行时中手动启动。手动启动启动异步任务一般都是通过运行时提供的.spawn()函数来完成。

1
2
3
4
5
6
7
8
9

#[tokio::main]
async fn main() {
  let handle = tokio::spawn(async {
    // 执行异步任务
  });

  // 如果需要从spawn的异步任务中获取返回值，就需要以下操作
  let result = handle.await.unwrap();
}

这种手动启动异步任务的方法很有用，可以跟多线程编程一样，创建仅会执行操作但不返回值的任务，例如定时任务等。

异步不适合处理什么

异步不是增强程序运行速度和并行容量的万灵药，异步也有一些场景是不合适合使用的。

  1. CPU密集型任务：CPU密集型任务会占用大量的CPU时间，无法通过出让时间片的方法使程序的并行处理能力得到加强。在处理这种类型的任务时一般只能通过开启多个线程来使每个CPU密集型任务充分利用CPU核心来加速。
  2. 读写大量文件的任务：虽然IO密集型任务非常适合使用异步来处理，但是大量文件读写的操作并不会因为程序采用了异步处理就会变得性能更好，这主要是受到了操作系统中文件系统的限制。
  3. 发送单个Web请求：异步运行时在响应Web请求的时候是可以通轮换活跃处理活动来使服务的并发能力提升的，但是在大宋单一Web请求的时候，程序能做的事情只有等待，所以在发送和处理单个Web请求的时候使用异步也不会有太大的性能提升效果。
索引标签
Rust
异步
Async/Await
Future