actix 最小通道示例

Question

我正在尝试在 actix-web 中复制异步通道的简单基准。然而，actix 似乎永远不会调用

recv

函数。怎么会这样？

发送线程中的

await

永远不会导致 tokio 屈服于其他线程吗？我以为

await

产生了当前上下文？

#[actix_web::main]
async fn main() {
    let (s, r) = async_channel::unbounded();

    let t1 = tokio::spawn(async move {
        let mut now = chrono::Utc::now();
        while let Ok(x) = r.recv().await {
            if x % 1 == 0 {
                let next_now = chrono::Utc::now();
                println!("{} {:?}", x, next_now - now);
                now = next_now;
            }
        }
    });

    let t2 = tokio::spawn(async move {
        for i in 1.. {
            s.send(i).await.unwrap();
        }
    });

    join!(t1, t2).await;
}

我尝试将每个代码块放在生成调用之外，并替换“加入！”与 select 和其他 future 调用（也完全摆脱它）。

我还试图弄清楚是否可以将 actix-web 与 tokio 一起使用（具体来说，在 actix-web 中使用 tokio::spawn ，在 tokio::main 中使用 actix::rt::spawn ）。

Answer 1

for i in 1.. {
    s.send(i).await.unwrap();
}

这个循环永远不会屈服。是的，那里有一个

await

，但是

send()

仅在通道已满时才会真正暂停。当您创建一个

unbounded()

通道时，它会在被认为已满之前耗尽内存。如果没有屈服，其他任务将永远不会被执行。

尝试将您的频道更改为

bounded

频道，或尝试在循环中

sleep

：

let t2 = tokio::spawn(async move {
    for i in 1.. {
        s.send(i).await.unwrap();
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
});

如果我必须选择，我会让通道有界。将无限生成器与无界通道相结合始终是一个坏主意。