|
|
|
|
@ -54,7 +54,7 @@ async fn main() {
|
|
|
|
|
let (tx2, rx2) = oneshot::channel();
|
|
|
|
|
|
|
|
|
|
tokio::spawn(async {
|
|
|
|
|
// 等待 `some_peration` 的完成
|
|
|
|
|
// 等待 `some_operation` 的完成
|
|
|
|
|
// 或者处理 `oneshot` 的关闭通知
|
|
|
|
|
tokio::select! {
|
|
|
|
|
val = some_operation() => {
|
|
|
|
|
@ -62,12 +62,10 @@ async fn main() {
|
|
|
|
|
}
|
|
|
|
|
_ = tx1.closed() => {
|
|
|
|
|
// 收到了发送端发来的关闭信号
|
|
|
|
|
//
|
|
|
|
|
// `select` 即将结束，此时，正在进行的 `some_operation()` 任务会被取消，任务自动完成， // tx1 被释放
|
|
|
|
|
// `select` 即将结束，此时，正在进行的 `some_operation()` 任务会被取消，任务自动完成，
|
|
|
|
|
// tx1 被释放
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
tokio::spawn(async {
|
|
|
|
|
@ -85,10 +83,10 @@ async fn main() {
|
|
|
|
|
}
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
上面代码的重点就在于 `tx1.closed` 所在的分支，一旦发送端被关闭，那该分支就会被执行，然后 `select` 会退出，并清理掉还没执行的第一个分支 `val = some_operation()` ，这其中 `some_peration` 返回的 `Future` 也会被清理，根据之前的内容，`Future` 被清理那相应的任务会立即取消，因此 `some_oeration` 会被取消，不再执行。
|
|
|
|
|
上面代码的重点就在于 `tx1.closed` 所在的分支，一旦发送端被关闭，那该分支就会被执行，然后 `select` 会退出，并清理掉还没执行的第一个分支 `val = some_operation()` ，这其中 `some_operation` 返回的 `Future` 也会被清理，根据之前的内容，`Future` 被清理那相应的任务会立即取消，因此 `some_operation` 会被取消，不再执行。
|
|
|
|
|
|
|
|
|
|
#### Future 的实现
|
|
|
|
|
为了更好的理解 `select` 的工作原理，我们来看看如果使用 `Future`。当然，这里是一个简化版本，在实际中，`select!` 会包含一些额外的功能，例如一开始会随机选择一个分支进行 `poll`。
|
|
|
|
|
为了更好的理解 `select` 的工作原理，我们来看看如果使用 `Future` 该如何实现。当然，这里是一个简化版本，在实际中，`select!` 会包含一些额外的功能，例如一开始会随机选择一个分支进行 `poll`。
|
|
|
|
|
|
|
|
|
|
```rust
|
|
|
|
|
use tokio::sync::oneshot;
|
|
|
|
|
@ -150,7 +148,7 @@ async fn main() {
|
|
|
|
|
|
|
|
|
|
当 `select` 宏开始执行后，所有的分支会开始并发的执行。当任何一个**表达式**完成时，会将结果跟**模式**进行匹配。若匹配成功，则剩下的表达式会被释放。
|
|
|
|
|
|
|
|
|
|
最常用的**模式**就是用变量名去匹配表达式返回的值，然乎该变量就可以在**结果处理**环节使用。
|
|
|
|
|
最常用的**模式**就是用变量名去匹配表达式返回的值，然后该变量就可以在**结果处理**环节使用。
|
|
|
|
|
|
|
|
|
|
如果当前的模式不能匹配，剩余的 `async` 表达式将继续并发的执行，直到下一个完成。
|
|
|
|
|
|
|
|
|
|
@ -245,7 +243,7 @@ async fn main() {
|
|
|
|
|
## 错误传播
|
|
|
|
|
在 Rust 中使用 `?` 可以对错误进行传播，但是在 `select!` 中，`?` 如何工作取决于它是在分支中的 `async` 表达式使用还是在结果处理的代码中使用:
|
|
|
|
|
|
|
|
|
|
- 在分支中 `ascyn` 表达式使用会将该表达式的结果变成一个 `Result`
|
|
|
|
|
- 在分支中 `async` 表达式使用会将该表达式的结果变成一个 `Result`
|
|
|
|
|
- 在结果处理中使用，会将错误直接传播到 `select!` 之外
|
|
|
|
|
|
|
|
|
|
```rust
|
|
|
|
|
@ -321,7 +319,7 @@ async fn main() {
|
|
|
|
|
上面代码中，`rx` 通道关闭后，`recv()` 方法会返回一个 `None`，可以看到没有任何模式能够匹配这个 `None`，那为何不会报错？秘密就在于 `else` 上：当使用模式去匹配分支时，若之前的所有分支都无法被匹配，那 `else` 分支将被执行。
|
|
|
|
|
|
|
|
|
|
## 借用
|
|
|
|
|
当在 Tokio 中生成( spawan )任务时，其 async 语句块必须拥有其中数据的所有权。而 `select!` 并没有这个限制，它的每个分支表达式可以直接借用数据，然后进行并发操作。只要遵循 Rust 的借用规则，多个分支表达式就可以不可变的借用同一个数据，或者在一个表达式可变的借用某个数据。
|
|
|
|
|
当在 Tokio 中生成( spawn )任务时，其 async 语句块必须拥有其中数据的所有权。而 `select!` 并没有这个限制，它的每个分支表达式可以直接借用数据，然后进行并发操作。只要遵循 Rust 的借用规则，多个分支表达式可以不可变的借用同一个数据，或者在一个表达式可变的借用某个数据。
|
|
|
|
|
|
|
|
|
|
来看个例子，在这里我们同时向两个 TCP 目标发送同样的数据:
|
|
|
|
|
```rust
|
|
|
|
|
@ -357,7 +355,7 @@ async fn race(
|
|
|
|
|
|
|
|
|
|
如果你把连接过程放在了结果处理中，那连接失败会直接从 `race` 函数中返回，而不是继续执行另一个分支中的连接！
|
|
|
|
|
|
|
|
|
|
还有一个非常重要的点，**借用规则在分支表达式和结果处理中存在很大的不同**。例如上面代码中，我们在两个分支表达式中分别对 `data` 做了不可变借用，这当然ok，但是若是两次可变借用，那编译器会立即进行报错。但是转折来了：当在结果处理中进行两次可变借用时，却不会报错，大家可以思考下为什么，提示下：思考下分支在执行成后会发生什么。
|
|
|
|
|
还有一个非常重要的点，**借用规则在分支表达式和结果处理中存在很大的不同**。例如上面代码中，我们在两个分支表达式中分别对 `data` 做了不可变借用，这当然ok，但是若是两次可变借用，那编译器会立即进行报错。但是转折来了：当在结果处理中进行两次可变借用时，却不会报错，大家可以思考下为什么，提示下：思考下分支在执行完成后会发生什么？
|
|
|
|
|
|
|
|
|
|
```rust
|
|
|
|
|
use tokio::sync::oneshot;
|
|
|
|
|
@ -540,7 +538,7 @@ async fn main() {
|
|
|
|
|
|
|
|
|
|
当第一次循环开始时， 第一个分支会立即完成，因为 `operation` 的参数是 `None`。当第一个分支执行完成时，`done` 会变成 `true`，此时第一个分支的条件将无法被满足，开始执行第二个分支。
|
|
|
|
|
|
|
|
|
|
当第二个分支收到一个偶数时，`done` 会被修改为 `false`，且 `operation` 被设置了值。 此后再一次循环时，第一个分支会被执行，且 `oertation` 返回一个 `Some(2)`，因此会触发 `return` ，最终结束循环并返回。
|
|
|
|
|
当第二个分支收到一个偶数时，`done` 会被修改为 `false`，且 `operation` 被设置了值。 此后再一次循环时，第一个分支会被执行，且 `operation` 返回一个 `Some(2)`，因此会触发 `return` ，最终结束循环并返回。
|
|
|
|
|
|
|
|
|
|
这段代码引入了一个新的语法: `if !done`，在解释之前，先看看去掉后会如何：
|
|
|
|
|
```console
|
|
|
|
|
@ -556,10 +554,10 @@ note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
|
|
|
|
|
|
|
|
|
|
那大家肯定有疑问了，既然 `operation` 不能再被调用了，我们该如何在有偶数值时，再回到第一个分支对其进行调用呢？答案就是 `operation.set(action(Some(v)));`，该操作会重新使用新的参数设置 `operation`。
|
|
|
|
|
|
|
|
|
|
## spawan 和 select! 的一些不同
|
|
|
|
|
## spawn 和 select! 的一些不同
|
|
|
|
|
学到现在，相信大家对于 `tokio::spawn` 和 `select!` 已经非常熟悉，它们的共同点就是都可以并发的运行异步操作。
|
|
|
|
|
然而它们使用的策略大相径庭。
|
|
|
|
|
|
|
|
|
|
`tokio::spawan` 函数会启动新的任务来运行一个异步操作，每个任务都是一个独立的对象可以单独被 Tokio 调度运行，因此两个不同的任务的调度都是独立进行的，甚至于它们可能会运行在两个不同的操作系统线程上。鉴于此，生成的任务和生成的线程有一个相同的限制：不允许对外部环境中的值进行借用。
|
|
|
|
|
`tokio::spawn` 函数会启动新的任务来运行一个异步操作，每个任务都是一个独立的对象可以单独被 Tokio 调度运行，因此两个不同的任务的调度都是独立进行的，甚至于它们可能会运行在两个不同的操作系统线程上。鉴于此，生成的任务和生成的线程有一个相同的限制：不允许对外部环境中的值进行借用。
|
|
|
|
|
|
|
|
|
|
而 `select!` 宏就不一样了，它在同一个任务中并发运行所有的分支，正是因为这样在同一个任务中，这些分支无法被同时运行。 `select!` 宏在单个任务中实现了多路复用的功能。
|
|
|
|
|
而 `select!` 宏就不一样了，它在同一个任务中并发运行所有的分支。正是因为这样，在同一个任务中，这些分支无法被同时运行。 `select!` 宏在单个任务中实现了多路复用的功能。
|
|
|
|
|
|
lijinpeng
3 years ago