如何从标准库生成的线程向 Tokio 异步任务发送消息？

Question

我有一个设置，其中我的程序使用 std::thread::spawn.

为 CPU 绑定计算生成多个线程

我需要一个 GRPC 服务器来处理传入的命令并流式传输工作线程完成的输出。我正在为 GRPC 服务器使用 tonic，它只在 Tokio future 中提供异步实现。

我需要能够将消息从我的“正常”标准库线程发送到 Tokio future。

我已将我的代码简化为此处的最低限度：

use std::thread;
use tokio::sync::mpsc; // 1.9.0

fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    let tokio_runtime = tokio::runtime::Runtime::new().unwrap();
    tokio_runtime.spawn(async move {
        // the code below starts the GRPC server in reality, here I'm just demonstrating trying to receive a message
        while let Some(v) = rx.recv().await {}
    });

    let h = thread::spawn(move || {
        // do work
        tx.send(1).await; //<------ error occurs here since I can't await in a non-async block
    });

    h.join().unwrap();
}

我的主工作线程如何与 Tokio 生成的 GRPC 服务器通信？

Answer 1

您可以使用 tokio 的 sync 功能。有两个选项 - UnboundedSender and Sender::blocking_send().

unbounded sender 的问题在于它没有背压，如果您的生产者比消费者更快，您的应用程序可能会因内存不足错误而崩溃或耗尽生产者使用的其他有限资源。

作为一般规则，您应该避免使用无界队列，这给我们留下了更好的选择 blocking_send():

Playground:

use std::thread;
use tokio::sync::mpsc; // 1.9.0

fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    let tokio_runtime = tokio::runtime::Runtime::new().unwrap();
    tokio_runtime.spawn(async move {
        // the code below starts the GRPC server in reality, here I'm just demonstrating trying to receive a message
        while let Some(v) = rx.recv().await {
            println!("Received: {:?}", v);
        }
    });

    let h = thread::spawn(move || {
        // do work
        tx.blocking_send(1).unwrap();
    });

    h.join().unwrap();
}

如何从标准库生成的线程向 Tokio 异步任务发送消息？

How can I send a message from a standard library spawned thread to a Tokio async task?

multithreading

rust

rust-tokio

rust-tonic