Tokio教程之通道

Posted 跨链技术践行者

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Tokio教程之通道相关的知识,希望对你有一定的参考价值。

通道

Tokio教程之通道

https://tokio.rs/tokio/tutorial/channels

假设我们想运行两个并发的Redis命令。我们可以为每个命令生成一个任务。那么这两条命令就会同时发生。

起初,我们可能会尝试类似的做法。

use mini_redis::client;

#[tokio::main]
async fn main() {
    // Establish a connection to the server
    let mut client = client::connect("127.0.0.1:6379").await.unwrap();

    // Spawn two tasks, one gets a key, the other sets a key
    let t1 = tokio::spawn(async {
        let res = client.get("hello").await;
    });

    let t2 = tokio::spawn(async {
        client.set("foo", "bar".into()).await;
    });

    t1.await.unwrap();
    t2.await.unwrap();
}

Copy

这不会被编译,因为两个任务都需要以某种方式访问 client。由于Client没有实现Copy,如果没有一些代码来促进这种共享,它将不会被编译。此外,Client::set 需要 &mut self,这意味着调用它需要独占访问。我们可以为每个任务打开一个连接,但这并不理想。我们不能使用 std::sync::Mutex,因为 .await 需要在持有锁的情况下被调用。我们可以使用 tokio::sync::Mutex,但这只允许一个飞行中的请求。如果客户端实现了管道化,那么异步Mutex会导致连接的利用率不足。

消息传递

答案是使用消息传递。这种模式包括产生一个专门的任务来管理 client 资源。任何希望发出请求的任务都会向 client 任务发送一个消息。client 任务代表发送者发出请求,并将响应发回给发送者。

使用这种策略,就可以建立一个单一的连接。管理 client 的任务能够获得排他性的访问,以便调用get和set。此外,通道作为一个缓冲区工作。在 client 任务忙碌的时候,操作可以被发送到 client 任务。一旦 client 任务可以处理新的请求,它就从通道中拉出下一个请求。这可以带来更好的吞吐量,并且可以扩展到支持连接池。

Tokio的通道原语

Tokio提供一些通道(channel),每个通道都有不同的用途:

  • mpsc:多生产者,单消费者通道。可以发送许多数值。
  • oneshot:单生产者,单消费者通道。可以发送一个单一的值。
  • broadcast: 多生产者,多消费者。可以发送许多值。每个接收者看到每个值。
  • watch:单生产者,多消费者。可以发送许多值,但不保留历史。接收者只看到最近的值。

如果你需要一个多生产者多消费者的通道,只有一个消费者看到每个消息,你可以使用 async-channel crate。也有一些通道用于异步Rust之外,比如 std::sync::mpsc 和crossbeam::channel。这些通道通过阻塞线程来等待消息,这在异步代码中是不允许的。

在本节中,我们将使用 mpsc 和 oneshot。其他类型的消息传递通道将在后面的章节中进行探讨。本节的完整代码可在此找到。

定义消息类型

在大多数情况下,当使用消息传递时,接收消息的任务会对一个以上的命令做出响应。在我们的例子中,该任务将对GET和SET命令做出响应。为了对此进行建模,我们首先定义一个Command枚举,并为每个命令类型包含一个变量。

use bytes::Bytes;

#[derive(Debug)]
enum Command {
    Get {
        key: String,
    },
    Set {
        key: String,
        val: Bytes,
    }
}	

Copy

创建通道

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Create a new channel with a capacity of at most 32.
    let (tx, mut rx) = mpsc::channel(32);

    // ... Rest comes here
}

Copy

mpsc 通道被用来向管理redis连接的任务发送命令。多生产者的能力允许从许多任务发送消息。创建通道会返回两个值,一个发送者和一个接收者。这两个句柄是单独使用的。它们可以被转移到不同的任务。

通道的创建容量为32。如果消息的发送速度比接收速度快,通道将储存这些消息。一旦通道中存储了32条消息,调用 send(...).await 将进入睡眠状态,直到有消息被接收者删除。

从多个任务中发送是通过 cloning sender 来完成的。比如说。

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    let tx2 = tx.clone();

    tokio::spawn(async move {
        tx.send("sending from first handle").await;
    });

    tokio::spawn(async move {
        tx2.send("sending from second handle").await;
    });

    while let Some(message) = rx.recv().await {
        println!("GOT = {}", message);
    }
}

Copy

两条信息都被发送到单一的 Receiver 句柄。不可能克隆一个 mpsc 通道的Receiver。

当每个 sender 都超出了范围或被放弃时,就不再可能向通道发送更多的消息了。在这一点上,接收器上的 recv 调用将返回None,这意味着所有的发送者都消失了,通道被关闭。

在我们管理 Redis 连接的任务中,它知道一旦通道关闭,它就可以关闭 Redis 连接,因为该连接将不会再被使用。

生成管理器任务

接下来,生成一个任务,处理来自通道的消息。首先,一个客户端连接被建立到Redis。然后,通过Redis连接发出收到的命令。

use mini_redis::client;
// The `move` keyword is used to **move** ownership of `rx` into the task.
let manager = tokio::spawn(async move {
    // Establish a connection to the server
    let mut client = client::connect("127.0.0.1:6379").await.unwrap();

    // Start receiving messages
    while let Some(cmd) = rx.recv().await {
        use Command::*;

        match cmd {
            Get { key } => {
                client.get(&key).await;
            }
            Set { key, val } => {
                client.set(&key, val).await;
            }
        }
    }
});

Copy

现在,更新这两个任务,通过通道发送命令,而不是直接在Redis连接上发布。

// The `Sender` handles are moved into the tasks. As there are two
// tasks, we need a second `Sender`.
let tx2 = tx.clone();

// Spawn two tasks, one gets a key, the other sets a key
let t1 = tokio::spawn(async move {
    let cmd = Command::Get {
        key: "hello".to_string(),
    };

    tx.send(cmd).await.unwrap();
});

let t2 = tokio::spawn(async move {
    let cmd = Command::Set {
        key: "foo".to_string(),
        val: "bar".into(),
    };

    tx2.send(cmd).await.unwrap();
});

Copy

在 main 函数的底部,我们 .await join 句柄,以确保在进程退出前完全完成命令。

t1.await.unwrap();
t2.await.unwrap();
manager.await.unwrap();

Copy

接收响应

最后一步是接收来自管理器任务的响应。GET命令需要获得数值,SET命令需要知道操作是否成功完成。

为了传递响应,使用了一个 oneshot 通道。oneshot 通道是一个单一生产者、单一消费者的通道,为发送单一数值而优化。在我们的例子中,这个单一的值就是响应。

与mpsc类似,oneshot::channel() 返回一个 sender 和 receiver 句柄。

use tokio::sync::oneshot;

let (tx, rx) = oneshot::channel();

Copy

与mpsc不同,没有指定容量,因为容量总是1。此外,两个句柄都不能被克隆。

为了接收来自管理任务的响应,在发送命令之前,创建了一个 oneshot 通道。该通道的 Sender 部分被包含在给管理任务的命令中。receive 部分用来接收响应。

首先,更新 Command 以包括 Sender。为了方便起见,使用一个类型别名来引用 Sender。

use tokio::sync::oneshot;
use bytes::Bytes;

/// Multiple different commands are multiplexed over a single channel.
#[derive(Debug)]
enum Command {
    Get {
        key: String,
        resp: Responder<Option<Bytes>>,
    },
    Set {
        key: String,
        val: Vec<u8>,
        resp: Responder<()>,
    },
}

/// Provided by the requester and used by the manager task to send
/// the command response back to the requester.
type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;

Copy

现在,更新发布命令的任务,包括 oneshot::Sender

let t1 = tokio::spawn(async move {
    let (resp_tx, resp_rx) = oneshot::channel();
    let cmd = Command::Get {
        key: "hello".to_string(),
        resp: resp_tx,
    };

    // Send the GET request
    tx.send(cmd).await.unwrap();

    // Await the response
    let res = resp_rx.await;
    println!("GOT = {:?}", res);
});

let t2 = tokio::spawn(async move {
    let (resp_tx, resp_rx) = oneshot::channel();
    let cmd = Command::Set {
        key: "foo".to_string(),
        val: b"bar".to_vec(),
        resp: resp_tx,
    };

    // Send the SET request
    tx2.send(cmd).await.unwrap();

    // Await the response
    let res = resp_rx.await;
    println!("GOT = {:?}", res);
});

Copy

最后,更新管理任务,通过 oneshot 的通道发送响应。

while let Some(cmd) = rx.recv().await {
    match cmd {
        Command::Get { key, resp } => {
            let res = client.get(&key).await;
            // Ignore errors
            let _ = resp.send(res);
        }
        Command::Set { key, val, resp } => {
            let res = client.set(&key, val.into()).await;
            // Ignore errors
            let _ = resp.send(res);
        }
    }
}

Copy

在 oneshot::Sender 上调用 send 会立即完成,不需要 .await。这是因为在 oneshot 通道上的发送将总是立即失败或成功,而不需要任何形式的等待。

在 oneshot 通道上发送一个值,当接收方的一半已经放弃时,返回 Err。这表明接收方对响应不再感兴趣了。在我们的方案中,接收方取消兴趣是一个可接受的事件。resp.send(...)返回的Err不需要被处理。

背压和有界通道

无论何时引入并发或队列,都必须确保队列是有界的,系统将优雅地处理负载。无界的队列最终会占用所有可用的内存,导致系统以不可预测的方式失败。

Tokio 小心避免隐性队列。这其中很大一部分是由于异步操作是 lazy 的。请考虑以下情况。

loop {
    async_op();
}

Copy

如果异步操作急切地运行,循环将重复排队运行一个新的 async_op,而不确保之前的操作完成。这就导致了隐性的无边界队列。基于回调的系统和基于急切的 future 的系统特别容易受此影响。

然而,在Tokio和异步Rust中,上述片段根本不会导致 async_op 的运行。这是因为 .await 从未被调用。如果该片段被更新为使用 .await,那么循环会等待操作完成后再重新开始。

loop {
    // Will not repeat until `async_op` completes
    async_op().await;
}

Copy

必须明确地引入并发和队列。做到这一点的方法包括:

  • tokio::spawn
  • select!
  • join!
  • mpsc::channel

在这样做的时候,要注意确保并发的总量是有界限的。例如,当编写一个TCP接受循环时,要确保打开的套接字的总数是有限制的。当使用 mpsc::channel 时,选择一个可管理的通道容量。具体的约束值将是特定于应用的。

注意和挑选好的界限是编写可靠的 tokio 应用程序的一个重要部分。

以上是关于Tokio教程之通道的主要内容,如果未能解决你的问题,请参考以下文章

Tokio教程之select

Tokio教程之stream

Tokio教程之深入异步

Tokio教程之分帧

Tokio教程之I/O

Hello Tokio