Tokio教程之桥接同步代码

Posted 2021-09-09 跨链技术践行者

桥接同步代码

Tokio教程之桥接同步代码

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::set
• Client::set_expires
• Client::publish
• 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。