rust - 如何使用 tokio 的 UdpSocket 处理 1 台服务器中的消息 : N clients setup?

Question

我想做的事:

...编写一个(1)个服务器/(N)个客户端(网络游戏)架构，它使用 UDP 套接字作为通信的底层基础。

消息发送为 Vec<u8> , 通过 bincode 编码( crate )

我还希望能够偶尔发送超过典型最大值 MTU 的数据报~ 1500 bytes并在接收端正确组装，包括发送ack -消息等 (我想我必须自己实现，对吧？)

对于 UdpSocket我考虑过使用 tokio的实现，也许 framed .我不确定这是否是一个好的选择，因为这似乎会引入不必要的映射步骤 Vec<u8> (由 bincode 序列化)到 Vec<u8> ( UdpCodec 中的 tokio 需要)(？)

考虑这个最小的代码示例:

Cargo.toml (服务器)

bincode = "1.0"
futures = "0.1"
tokio-core = "^0.1"

( Serde 和 serde-derive 用于定义协议(protocol)的 shared crate !)

(我想尽快将 tokio-core 替换为 tokio)

fn main() -> () {
    let addr = format!("127.0.0.1:{port}", port = 8080);
    let addr = addr.parse::<SocketAddr>().expect(&format!("Couldn't create valid SocketAddress out of {}", addr));

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let socket = UdpSocket::bind(&addr, &handle).expect(&format!("Couldn't bind socket to address {}", addr));


    let udp_future = socket.framed(MyCodec {}).for_each(|(addr, data)| {
        socket.send_to(&data, &addr); // Just echo back the data
        Ok(())
    });

    core.run(udp_future).unwrap();
}

struct MyCodec;

impl UdpCodec for MyCodec {
    type In = (SocketAddr, Vec<u8>);
    type Out = (SocketAddr, Vec<u8>);

    fn decode(&mut self, src: &SocketAddr, buf: &[u8]) -> io::Result<Self::In> {
        Ok((*src, buf.to_vec()))
    }

    fn encode(&mut self, msg: Self::Out, buf: &mut Vec<u8>) -> SocketAddr {
        let (addr, mut data) = msg;
        buf.append(&mut data);
        addr
    }
}

这里的问题是:

let udp_future = socket.framed(MyCodec {}).for_each(|(addr, data)| { | ------ value moved here ^^^^^^^^^^^^^^ value captured here after move | = note: move occurs because socket has type tokio_core::net::UdpSocket, which does not implement the Copy trait

该错误完全有道理，但我不确定如何创建如此简单的回显服务。实际上，消息的处理涉及到更多的逻辑，但是为了一个最小的例子，这应该足以给出一个粗略的想法。

我的解决方法是一个丑陋的黑客:创建第二个套接字。

Answer 1

这是 UdpSocket::framed 的签名来自 Tokio 的文档:

pub fn framed<C: UdpCodec>(self, codec: C) -> UdpFramed<C>

请注意，它需要 self ，而不是 &self ;也就是说，调用这个函数会消耗套接字。 UdpFramed当您调用它时，包装器拥有底层套接字。您的编译错误告诉您正在移动 socket当你调用这个方法，但你也在尝试借用 socket在您的闭包内(调用 send_to )。

这可能不是您想要的真实代码。使用的全部要点 framed()是将您的套接字变成更高级别的东西，因此您可以直接发送编解码器的项目，而不必组装数据报。使用 send或 send_to直接在套接字上可能会破坏您的消息协议(protocol)的框架。在这段代码中，你试图实现一个简单的回显服务器，你不需要使用 framed一点也不。但是，如果您确实想吃蛋糕并同时使用 framed和 send_to ，幸运的是 UdpFramed仍然允许您借入标的 UdpSocket , 使用 get_ref .您可以通过以下方式解决您的问题:

let framed = {
    let socket = UdpSocket::bind(&addr, &handle).expect(&format!("Couldn't bind socket to address {}", addr));
    socket.framed(MyCodec {})
}

let udp_future = framed.for_each(|(addr, data)| {
    info!(self.logger, "Udp packet received from {}: length: {}", addr, data.len());
    framed.get_ref().send_to(&data, &addr); // Just echo back the data

    Ok(())
});

我没有检查过这段代码，因为(正如 Shepmaster 正确指出的那样)你的代码片段还有其他问题，但无论如何它应该会给你这个想法。我将重复我之前的警告:如果您在实际代码中执行此操作，它将破坏您正在使用的网络协议(protocol)。 get_ref的文档是这样写的:

Note that care should be taken to not tamper with the underlying stream of data coming in as it may corrupt the stream of frames otherwise being worked with.

要回答您问题的新部分:是的，您需要自己处理重组，这意味着您的编解码器确实需要对您发送的字节进行一些帧处理。通常，这可能涉及 Vec<u8> 中不能出现的开始序列。 .开始序列让您在数据包丢失后识别下一条消息的开始(这在 UDP 中经常发生)。如果 Vec<u8> 中没有不能出现的字节序列，当它发生时你需要逃避它。然后您可以发送消息的长度，然后是数据本身；或者只是数据，后跟一个结束序列和一个校验和，这样你就知道没有丢失。这些设计有利有弊，这本身就是一个很大的话题。

您还需要您的 UdpCodec包含数据:来自 SocketAddr 的 map 到当前正在进行的部分重组消息。在 decode ，如果给你消息的开头，将其复制到 map 中并返回 Ok .如果给定消息的中间，并且映射中已经有消息的开头(对于那个 SocketAddr )，则将缓冲区附加到现有缓冲区并返回 Ok .当您到达消息的末尾时，返回整个内容并清空缓冲区。 UdpCodec 上的方法拍 &mut self为了启用这个用例。 ( 注意 理论上，您还应该处理乱序到达的数据包，但这在现实世界中实际上很少见。)
encode简单得多:您只需要添加相同的框架并将消息复制到缓冲区中。

让我在这里重申一下，在调用 framed() 之后，您不需要也不应该使用底层套接字。在上面。 UdpFramed既是源又是接收器，因此您也可以使用该对象来发送回复。您甚至可以使用 split()分开 Stream和 Sink如果这使您的应用程序中的所有权更容易，则可以从中实现。

总的来说，现在我已经看到了你正在努力解决的问题，我建议只使用几个 TCP 套接字而不是 UDP。如果您想要一个面向连接的可靠协议(protocol)，TCP 已经存在并且可以为您做到这一点。很容易花大量时间在 UDP 之上创建一个“可靠”层，它比 TCP 慢且不可靠。