gpt4 book ai didi

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

转载 作者:行者123 更新时间:2023-12-05 02:40:30 24 4
gpt4 key购买 nike

我有一个设置,我的程序使用 std::thread::spawn 为 CPU 绑定(bind)计算生成多个线程。

我需要一个 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 服务器通信?

最佳答案

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

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

作为一般规则,您应该避免使用无界队列,这让我们有了使用 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();
}

关于multithreading - 如何从标准库生成的线程向 Tokio 异步任务发送消息?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/68561218/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com