Rust Async 并发编程:任务、消息传递与 `join`

发布于:2025-03-03 ⋅ 阅读:(25) ⋅ 点赞:(0)

1. 创建异步任务

在传统的多线程模型中,我们使用 std::thread::spawn 来创建新的线程。而在 async 模型中,使用 spawn_task 代替 thread::spawn 来创建异步任务,并结合 await 关键字来处理异步操作。

示例:使用 spawn_task 进行并发计算

use trpl::{spawn_task, sleep};
use std::time::Duration;

#[tokio::main]
async fn main() {
    let handle = spawn_task(async {
        for i in 1..=5 {
            println!("hi number {} from the second task!", i);
            sleep(Duration::from_millis(500)).await;
        }
    });

    for i in 1..=5 {
        println!("hi number {} from the first task!", i);
        sleep(Duration::from_millis(500)).await;
    }

    handle.await.unwrap();
}

运行结果(实际输出顺序可能不同):

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
...

在该示例中,我们创建了一个新的异步任务 spawn_task,并在主任务中执行另一个循环。因为 sleepasync 版本的,所以 await 允许其他任务在 sleep 期间继续执行,而不会阻塞整个线程。

2. 使用 join 让多个任务同时运行

当我们想要同时运行多个 async 任务并等待它们全部完成时,可以使用 trpl::join。它类似于 std::thread::JoinHandle::join(),但适用于 async 任务。

示例:使用 trpl::join 运行两个异步任务

use trpl::{join, sleep};
use std::time::Duration;

#[tokio::main]
async fn main() {
    let fut1 = async {
        for i in 1..=5 {
            println!("Task 1: {}", i);
            sleep(Duration::from_millis(500)).await;
        }
    };

    let fut2 = async {
        for i in 1..=5 {
            println!("Task 2: {}", i);
            sleep(Duration::from_millis(500)).await;
        }
    };

    join!(fut1, fut2).await;
}

输出顺序保持一致

Task 1: 1
Task 2: 1
Task 1: 2
Task 2: 2
...

相比于 spawn_tasktrpl::join 提供了更加确定的执行顺序,因为 join! 会公平地轮询任务,使得它们不会出现一个任务远远领先于另一个的情况。

3. 通过消息通道在任务间传递数据

std::mpsc::channel 类似,Rust 也提供了 trpl::channel 来进行异步消息传递。这种方式避免了共享内存带来的数据竞争问题。

示例:异步消息传递

use trpl::{channel, sleep};
use std::time::Duration;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = channel();

    let sender = async move {
        let messages = vec!["Hello", "from", "the", "async", "world!"];
        for msg in messages {
            tx.send(msg).await.unwrap();
            sleep(Duration::from_millis(500)).await;
        }
    };

    let receiver = async move {
        while let Some(msg) = rx.recv().await {
            println!("Received: {}", msg);
        }
    };

    trpl::join(sender, receiver).await;
}

输出

Received: Hello
Received: from
Received: the
Received: async
Received: world!

任务自动关闭

上面的代码能正确终止,因为 txasync move 块结束时会被 drop,从而让 rx.recv() 返回 None,结束 while let 循环。

如果我们希望多个任务发送数据,只需 clone 发送端:

let tx1 = tx.clone();
let sender1 = async move {
    tx1.send("Extra message").await.unwrap();
};

使用 trpl::join3 让多个任务并发执行。

4. 总结

  • spawn_task 用于创建新的异步任务。
  • await 允许任务在等待时释放 CPU 资源,提高并发性能。
  • trpl::join 确保多个任务公平执行。
  • trpl::channel 实现了异步消息传递,避免了数据竞争。
  • async move 确保所有权正确传递,避免任务执行过程中 tx 被持有过久导致 rx.recv() 无限等待。

Rust 的异步编程提供了强大的并发能力,同时通过 await 避免了传统多线程编程的阻塞问题,使得代码更高效、更易读。希望本文能帮助你更好地理解 async 并发编程的核心概念!