如何将 Tokio 设置为多线程 UDP 服务器？

Question

到目前为止，我发现的唯一示例是 Tokio 存储库中的单线程示例 Echo UDP。如何使用 Tokio 启动一个循环，该循环将产生一个新线程来处理新的 UDP 连接。

Answer 1

original version of the question的答案

如何设置 Tokio 0.2 以“监听”UDP 数据？

use tokio::net::UdpSocket; // "0.2.20", features = ["full"]

type Error = Box<dyn std::error::Error>;
type Result<T, E = Error> = std::result::Result<T, E>;

#[tokio::main]
async fn main() -> Result<()> {
    let mut socket = UdpSocket::bind("127.0.0.1:9999").await?;

    loop {
        let mut data = [0; 1024];
        let valid_bytes = socket.recv(&mut data).await?;
        let data = &data[..valid_bytes];

        eprintln!("Read {} bytes", data.len());
    }
}

（Tokio 1.4 的代码实际上是相同的，只需删除 mut 限定符。）

在一个窗口中：

% cargo run
Read 6 bytes
Read 5 bytes
Read 6 bytes

另一个：

% nc -u 127.0.0.1 9999
alpha
beta
gamma

current version of the question 的答案

如何将 Tokio 设置为多线程 UDP 服务器？

上面的代码是多线程的； Tokio is multithreaded by default。您可能正在寻求创建并发（可能是并行）工作；这可以通过生成任务来完成：

• task::spawn
• task::spawn_local
• task::spawn_blocking

Answer 2

要在 tokio 中创建可调度单元，您应该使用 tokio::task::spawn。如果底层运行时是多线程的，那么这些单元将由多个线程完成。

您可以通过在示例中添加几行来了解它是如何工作的

fn main() {
...
    let jh = tokio::task::spawn(server.run());
    println!("udp server started {:?}", std::thread::current().id());
    jh.await?;
...
}
fn run
... 
   loop {
        if let Some((size, peer)) = to_send {
            let amt = socket.send_to(&buf[..size], &peer).await?;
            println!("eched back {:?}", std::thread::current().id());
        }

        to_send = Some(socket.recv_from(&mut buf).await?);
        println!("read some stuff {:?}", std::thread::current().id());
    }