Tokio教程之桥接同步代码

阳光穿透心脏的1/2处 2022-09-07 12:25 139阅读 0赞

# 桥接同步代码 #

Tokio教程之桥接同步代码

> [https://tokio.rs/tokio/tutorial/bridging][https_tokio.rs_tokio_tutorial_bridging]

在我们到目前为止看到的例子中，我们用 `#[tokio::main]` 标记了主函数，并使整个项目成为异步的。然而，这对所有项目来说都是不可取的。例如，一个GUI应用程序可能希望在主线程上运行GUI代码，并在另一个线程上运行Tokio运行时。

本页解释了如何将异步/等待隔离到项目的一小部分。

## `#[tokio::main]`扩展到什么？ ##

`#[tokio::main]` 宏是一个将你的主函数替换为非同步主函数的宏，它启动一个运行时，然后调用你的代码。例如，这样的main函数

#[tokio::main]
    async fn main() {
        println!("Hello world");
    }

Copy

通过这个宏变成了这样:

fn main() {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async {
                println!("Hello world");
            })
    }

Copy

为了在我们自己的项目中使用 `async/await`，我们可以做一些类似的事情，我们利用. `block_on` 方法在适当的时候进入异步环境。

## 到mini-redis的同步接口 ##

在本节中，我们将讨论如何通过存储 Runtime 对象并使用其 block\_on 方法来构建 mini-redis 的同步接口。在接下来的章节中，我们将讨论一些替代的方法，以及何时应该使用每种方法。

我们要封装的接口是异步的 client 类型。它有几个方法，我们将实现以下方法的阻塞版本：

*  [`Client::get`][Client_get]
 *  [`Client::set`][Client_set]
 *  [`Client::set_expires`][Client_set_expires]
 *  [`Client::publish`][Client_publish]
 *  [`Client::subscribe`][Client_subscribe]

为此，我们引入了一个名为 `src/blocking_client.rs` 的新文件，并用一个围绕 async Client 类型的封装结构来初始化它:

use tokio::net::ToSocketAddrs;
    use tokio::runtime::Runtime;
    
    pub use crate::client::Message;
    
    /// Established connection with a Redis server.
    pub struct BlockingClient {
        /// The asynchronous `Client`.
        inner: crate::client::Client,
    
        /// A `current_thread` runtime for executing operations on the
        /// asynchronous client in a blocking manner.
        rt: Runtime,
    }
    
    pub fn connect<T: ToSocketAddrs>(addr: T) -> crate::Result<BlockingClient> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
    
        // Call the asynchronous connect method using the runtime.
        let inner = rt.block_on(crate::client::connect(addr))?;
    
        Ok(BlockingClient { inner, rt })
    }

Copy

在这里，我们把构造函数作为第一个例子，说明如何在非同步上下文中执行异步方法。我们使用Tokio Runtime类型上的block\_on方法来做到这一点，它执行一个异步方法并返回其结果。

一个重要的细节是对 current\_thread 运行时的使用。通常在使用Tokio时，你会使用默认的 multi\_thread 运行时，它将产生一堆后台线程，这样它就可以有效地同时运行许多东西。对于我们的用例，我们每次只做一件事，所以我们不会因为运行多个线程而获得任何好处。这使得 current\_thread 运行时成为完美的选择，因为它不会产生任何线程。

`enable_all` 调用启用了 Tokio 运行时的IO和定时器驱动。如果它们没有被启用，运行时就无法执行IO或定时器。

> 因为 current\_thread 运行时不产生线程，所以它只在 block\_on 被调用时运行。一旦 block\_on 返回，所有在该运行时上生成的任务将冻结，直到你再次调用 block\_on。如果催生的任务在不调用 block\_on 时必须继续运行，请使用 multi\_threaded runtime。

一旦我们有了这个结构体，大部分的方法就很容易实现:

use bytes::Bytes;
    use std::time::Duration;
    
    impl BlockingClient {
        pub fn get(&mut self, key: &str) -> crate::Result<Option<Bytes>> {
            self.rt.block_on(self.inner.get(key))
        }
    
        pub fn set(&mut self, key: &str, value: Bytes) -> crate::Result<()> {
            self.rt.block_on(self.inner.set(key, value))
        }
    
        pub fn set_expires(
            &mut self,
            key: &str,
            value: Bytes,
            expiration: Duration,
        ) -> crate::Result<()> {
            self.rt.block_on(self.inner.set_expires(key, value, expiration))
        }
    
        pub fn publish(&mut self, channel: &str, message: Bytes) -> crate::Result<u64> {
            self.rt.block_on(self.inner.publish(channel, message))
        }
    }

Copy

Client::subscribe方法更有趣，因为它将 Client 转化为 Subscriber 对象。我们可以用以下方式实现它:

/// A client that has entered pub/sub mode.
    ///
    /// Once clients subscribe to a channel, they may only perform
    /// pub/sub related commands. The `BlockingClient` type is
    /// transitioned to a `BlockingSubscriber` type in order to
    /// prevent non-pub/sub methods from being called.
    pub struct BlockingSubscriber {
        /// The asynchronous `Subscriber`.
        inner: crate::client::Subscriber,
    
        /// A `current_thread` runtime for executing operations on the
        /// asynchronous client in a blocking manner.
        rt: Runtime,
    }
    
    impl BlockingClient {
        pub fn subscribe(self, channels: Vec<String>) -> crate::Result<BlockingSubscriber> {
            let subscriber = self.rt.block_on(self.inner.subscribe(channels))?;
            Ok(BlockingSubscriber {
                inner: subscriber,
                rt: self.rt,
            })
        }
    }
    
    impl BlockingSubscriber {
        pub fn get_subscribed(&self) -> &[String] {
            self.inner.get_subscribed()
        }
    
        pub fn next_message(&mut self) -> crate::Result<Option<Message>> {
            self.rt.block_on(self.inner.next_message())
        }
    
        pub fn subscribe(&mut self, channels: &[String]) -> crate::Result<()> {
            self.rt.block_on(self.inner.subscribe(channels))
        }
    
        pub fn unsubscribe(&mut self, channels: &[String]) -> crate::Result<()> {
            self.rt.block_on(self.inner.unsubscribe(channels))
        }
    }

Copy

因此，subscribe 方法将首先使用运行时将异步 Client 转化为异步 Subscriber。然后，它将把产生的 Subscriber 与运行时一起存储，并使用 block\_on 实现各种方法。

请注意，异步 Subscriber 结构体有一个名为 get\_subscribed 的非异步方法。为了处理这个问题，我们只需直接调用它而不涉及运行时。

## 其他方法 ##

上节解释了实现同步包装器的最简单的方法，但这并不是唯一的方法。这些方法是：

*  创建一个 Runtime 并在异步代码上调用 block\_on。
 *  创建一个 Runtime 并在其上生成事物。
 *  在一个单独的线程中运行 Runtime 并向其发送消息。

我们已经看到了第一种方法。其他两种方法概述如下。

### 在运行时上生成东西 ###

运行时对象有一个叫做 spawn 的方法。当你调用这个方法时，你会创建一个新的在运行时上运行的后台任务。比如说

use tokio::runtime::Builder;
    use tokio::time::{sleep, Duration};
    
    fn main() {
        let runtime = Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
    
        let mut handles = Vec::with_capacity(10);
        for i in 0..10 {
            handles.push(runtime.spawn(my_bg_task(i)));
        }
    
        // Do something time-consuming while the background tasks execute.
        std::thread::sleep(Duration::from_millis(750));
        println!("Finished time-consuming task.");
    
        // Wait for all of them to complete.
        for handle in handles {
            // The `spawn` method returns a `JoinHandle`. A `JoinHandle` is
            // a future, so we can wait for it using `block_on`.
            runtime.block_on(handle).unwrap();
        }
    }
    
    async fn my_bg_task(i: u64) {
        // By subtracting, the tasks with larger values of i sleep for a
        // shorter duration.
        let millis = 1000 - 50 * i;
        println!("Task {} sleeping for {} ms.", i, millis);
    
        sleep(Duration::from_millis(millis)).await;
    
        println!("Task {} stopping.", i);
    }

Copy

在上面的例子中，我们在运行时催生了 10 个后台任务，然后等待所有的任务。作为一个例子，这可能是在图形应用程序中实现后台网络请求的一个好方法，因为网络请求在GUI主线程上运行太耗时了。相反，你在后台运行的 Tokio 运行时上生成请求，并让任务在请求完成后将信息送回GUI代码，如果你想要一个进度条，甚至可以递增。

在这个例子中，重要的是，运行时被配置为多线程运行时。如果你把它改为 current\_thread 运行时，你会发现耗时的任务会在任何后台任务开始之前完成。这是因为在 current\_thread 运行时上生成的后台任务只在调用 block\_on 时执行，否则运行时就没有地方可以运行它们。

这个例子通过对调用 spawn 返回的 JoinHandle 调用 block\_on 来等待生成的任务完成，但这并不是唯一的方法。这里有一些替代方法。

*  使用消息传递通道，如 `tokio::sync::mpsc`。
 *  修改由 Mutex 等保护的共享值。这对GUI中的进度条来说是一个很好的方法，GUI在每一帧都会读取共享值。

spawn 方法在 Handle 类型上也是可用的。Handle 类型可以被克隆，以便在一个运行时中获得许多句柄，每个 Handle 都可以用来在运行时中生成新的任务。

### 发送消息 ###

第三种技术是催生一个运行时，并使用消息传递来与之通信。这比其他两种方法涉及更多的模板，但它是最灵活的方法。你可以在下面找到一个基本的例子。

use tokio::runtime::Builder;
    use tokio::sync::mpsc;
    
    pub struct Task {
        name: String,
        // info that describes the task
    }
    
    async fn handle_task(task: Task) {
        println!("Got task {}", task.name);
    }
    
    #[derive(Clone)]
    pub struct TaskSpawner {
        spawn: mpsc::Sender<Task>,
    }
    
    impl TaskSpawner {
        pub fn new() -> TaskSpawner {
            // Set up a channel for communicating.
            let (send, mut recv) = mpsc::channel(16);
    
            // Build the runtime for the new thread.
            //
            // The runtime is created before spawning the thread
            // to more cleanly forward errors if the `unwrap()`
            // panics.
            let rt = Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
    
            std::thread::spawn(move || {
                rt.block_on(async move {
                    while let Some(task) = recv.recv().await {
                        tokio::spawn(handle_task(task));
                    }
    
                    // Once all senders have gone out of scope,
                    // the `.recv()` call returns None and it will
                    // exit from the while loop and shut down the
                    // thread.
                });
            });
    
            TaskSpawner {
                spawn: send,
            }
        }
    
        pub fn spawn_task(&self, task: Task) {
            match self.spawn.blocking_send(task) {
                Ok(()) => {},
                Err(_) => panic!("The shared runtime has shut down."),
            }
        }
    }

Copy

这个例子可以用很多方式来配置。例如，你可以使用一个 Semaphore 来限制活动任务的数量，或者你可以使用一个相反方向的通道来向 spawner 发送响应。当你以这种方式催生一个运行时，它是一种 actor。

[https_tokio.rs_tokio_tutorial_bridging]: https://tokio.rs/tokio/tutorial/bridging
[Client_get]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.get
[Client_set]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.set
[Client_set_expires]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.set_expires
[Client_publish]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.publish
[Client_subscribe]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.subscribe