基于网络协议

• 基于网络协议
  • 回顾网络协议
  • Rust生态对网络协议的支持
  • std::net
    • 服务端：TcpListener
    • 客户端：TcpStream
  • 处理网络连接的一般方法
    • 问题一：如何处理大量连接？
    • 问题二：如何处理共享信息？
  • 处理网络数据的一般方法
    • JSON序列化
    • 使用 protobuf 自定义协议
      • 如何界定一个消息帧（frame）
      • tokio/tokio_util
  • Tokio
  • 尝试改写
  • 总结：如何用 Rust 做基于 TCP 的网络开发

在互联网时代，谈到网络开发，我们想到的首先是 Web 开发以及涉及的部分 HTTP 协议和 WebSocket 协议。

所以对于网络开发，这个非常宏大的议题，当然是不可能、也没有必要覆盖全部内容的，这里按照如下顺序进行：

1. 先简单聊聊网络开发的大全景图
2. 然后重点学习如何使用 Rust 标准库以及生态系统中的库来做网络处理，包括网络连接、网络数据处理的一些方法
3. 最后也会介绍几种典型的网络通讯模型的使用。

回顾网络协议

七层模型中：

1. 链路层和网络层一般构建在操作系统之中，我们并不需要直接触及
2. 而表现层和应用层关系紧密
3. 所以在实现过程中，大部分应用程序只关心网络层、传输层和应用层：

Rust生态对网络协议的支持

tokio::net 提供了和 std::net 几乎一致的封装，一旦你熟悉了 std::net，tokio::net 里的功能对你来说都并不陌生。所以先从 std::net 开始了解。

std::net

std::net 下提供了处理 TCP / UDP 的数据结构，以及一些辅助结构：

• TCP：TcpListener / TcpStream，处理服务器的监听以及客户端的连接

• UDP：UdpSocket，处理 UDP socket

• 其它：IpAddr 是 IPv4 和 IPv6 地址的封装；SocketAddr，表示 IP 地址 + 端口的数据结构

这里就主要介绍一下 TCP 的处理（TcpListener/TcpStream），顺带会使用到 IpAddr / SocketAddr。

服务端：TcpListener

如果要创建一个 TCP server，我们可以:

1. 绑定监听：使用 TcpListener 绑定某个端口
2. 循环处理♻️：用 loop 循环处理接收到的客户端请求
3. 接收到请求后，会得到一个 TcpStream
4. 它实现了 Read / Write trait，可以像读写文件一样，进行 socket 的读写：

use std::{
    io::{Read, Write},
    net::TcpListener,
    thread,
};

fn main() {
    let listener = TcpListener::bind("0.0.0.0:9527").unwrap();
    loop {
        // unwrap(): 
        let (mut stream, addr) = listener.accept().unwrap();
        println!("Accepted a new connection: {}", addr);
        thread::spawn(move || {
            let mut buf = [0u8; 12];
            stream.read_exact(&mut buf).unwrap();
            println!("data: {:?}", String::from_utf8_lossy(&buf));
            // 一共写了 17 个字节
            stream.write_all(b"glad to meet you!").unwrap();
        });
    }
}

客户端：TcpStream

对于客户端，我们可以用 TcpStream::connect() 得到一个 TcpStream。

一旦客户端的请求被服务器接受，就可以发送或者接收数据：

use std::{
    io::{Read, Write},
    net::TcpStream,
};

fn main() {
    let mut stream = TcpStream::connect("127.0.0.1:9527").unwrap();
    // 一共写了 12 个字节
    stream.write_all(b"hello world!").unwrap();

    let mut buf = [0u8; 17];
    stream.read_exact(&mut buf).unwrap();
    println!("data: {:?}", String::from_utf8_lossy(&buf));
}

在这个例子中:

1. 客户端在连接成功后，会发送 12 个字节的 “hello world!“给服务器
2. 服务器读取并回复后，客户端会尝试接收完整的、来自服务器的 17 个字节的 “glad to meet you!”。

但是，目前客户端和服务器都需要硬编码要接收数据的大小，这样不够灵活，后续我们会看到如何通过使用消息帧（frame）更好地处理。

处理网络连接的一般方法

如果你使用某个 Web Framework 处理 Web 流量，那么无需关心网络连接，框架会帮你打点好一切，你只需要关心某个路由或者某个 RPC 的处理逻辑就可以了。

但如果你要在 TCP 之上构建自己的协议，那么你需要认真考虑如何妥善处理网络连接。

fn main() {
    ...
    loop {
        let (mut stream, addr) = listener.accept().unwrap();
        println!("Accepted a new connection: {}", addr);
        thread::spawn(move || {
            ...
        });
    }
}

我们来详细看看。

问题一：如何处理大量连接？

问题二：如何处理共享信息？

但使用锁，就意味着一旦在关键路径上需要访问被锁住的资源，整个系统的吞吐量都会受到很大的影响。

处理网络数据的一般方法

我们再来看看如何处理网络数据。

大部分时候，我们可以使用已有的应用层协议来处理网络数据，比如 HTTP。

JSON序列化

在 HTTP 协议下，基本上使用 JSON 构建 REST API / JSON API 是业界的共识，客户端和服务器也有足够好的生态系统来支持这样的处理。

你只需要使用 serde 让你定义的 Rust 数据结构具备 Serialize/Deserialize 的能力，然后用 serde_json 生成序列化后的 JSON 数据。

Rocket 是 Rust 的一个全功能的 Web 框架，类似于 Python 的 Django。

下面是一个使用 rocket 来处理 JSON 数据的例子。

rocket = { version = "0.5.0-rc.1", features = ["json"] }

#[macro_use]
extern crate rocket;

use rocket::serde::json::Json;
use rocket::serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
#[serde(crate = "rocket::serde")]
struct Hello {
    name: String,
}

#[get("/", format = "json")]
fn hello() -> Json<Hello> {
    Json(Hello { name: "Tyr".into() })
}

#[launch]
fn rocket() -> _ {
    rocket::build().mount("/", routes![hello])
}

可以看到，使用 rocket，10 多行代码，我们就可以运行起一个 Web Server。

如果你出于性能或者其他原因，可能需要定义自己的客户端 / 服务器间的协议。那么，可以使用传统的 TLV（Type-Length-Value）来描述协议数据，或者使用更加高效简洁的 protobuf。

使用 protobuf 自定义协议

protobuf 是一种非常方便的定义向后兼容协议的工具，它不仅能使用在构建 gRPC 服务的场景，还能用在其它网络服务中。

在之前的小项目中，无论是 thumbor 的实现，还是 kv server 的实现，都用到了 protobuf。

kv server 的实现在 TCP 之上构建了基于 protobuf 的协议，支持一系列 HXXX 命令。

不过，使用 protobuf 构建协议消息的时候需要注意，因为 protobuf 生成的是不定长消息，所以需要在客户端和服务器之间约定好：

如何界定一个消息帧（frame）

tokio/tokio_util

下面是使用 tokio / tokio_util 撰写的服务器和客户端，你可以看到，服务器和客户端都使用了 LengthDelimitedCodec 来处理消息帧。

use anyhow::Result;
use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use tokio::net::TcpListener;
use tokio_util::codec::{Framed, LengthDelimitedCodec};

#[tokio::main]
async fn main() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9527").await?;
    loop {
        let (stream, addr) = listener.accept().await?;
        println!("accepted: {:?}", addr);
        // LengthDelimitedCodec 默认 4 字节长度
        let mut stream = Framed::new(stream, LengthDelimitedCodec::new());

        tokio::spawn(async move {
            // 接收到的消息会只包含消息主体（不包含长度）
            while let Some(Ok(data)) = stream.next().await {
                println!("Got: {:?}", String::from_utf8_lossy(&data));
                // 发送的消息也脏需要发送消息主体，不需要提供长度
                // Framed/LengthDelimitedCodec 会自动计算并添加
                stream.send(Bytes::from("goodbye world!")).await.unwrap();
            }
        });
    }
}

use anyhow::Result;
use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use tokio::net::TcpStream;
use tokio_util::codec::{Framed, LengthDelimitedCodec};

#[tokio::main]
async fn main() -> Result<()> {
    let stream = TcpStream::connect("127.0.0.1:9527").await?;
    let mut stream = Framed::new(stream, LengthDelimitedCodec::new());
    stream.send(Bytes::from("hello world")).await?;

    // 接收从服务器返回的数据
    if let Some(Ok(data)) = stream.next().await {
        println!("Got: {:?}", String::from_utf8_lossy(&data));
    }

    Ok(())
}

和前面std::net的 TcpListener / TcpStream 代码相比:

双方都不需要知道对方发送的数据的长度，就可以通过 StreamExt trait 的 next() 接口 得到下一个消息；

• 在发送时，只需要调用 SinkExt trait 的 send() 接口 发送，相应的长度就会被自动计算并添加到要发送的消息帧的开头。

当然啦，如果你想自己运行这两段代码,

#![allow(unused)]
fn main() {
[dependencies]
anyhow = "1"
bytes = "1"
futures = "0.3"
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.6", features = ["codec"] }
}

这里为了代码的简便，并没有直接使用 protobuf。你可以把发送和接收到的 Bytes 里的内容视作 protobuf 序列化成的二进制（如果你想看 protobuf 的处理，可以看thumbor 和 kv server 的源代码）。

我们可以看到，使用 LengthDelimitedCodec，构建一个自定义协议，变得非常简单。短短二十行代码就完成了非常繁杂的工作。

Tokio

绝大多数情况下，我们应该使用支持异步的网络开发，所以你会在各种网络相关的代码中，看到 tokio 的身影。作为 Rust 下主要的异步网络运行时，你可以多花点时间了解它的功能。

尝试改写

在kv server 的 examples 里，使用过async_prost，

use anyhow::Result;
use async_prost::AsyncProstStream;
use futures::prelude::*;
use kv1::{CommandRequest, CommandResponse, Service, ServiceInner, SledDb};
use tokio::net::TcpListener;
use tracing::info;

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    let service: Service<SledDb> = ServiceInner::new(SledDb::new("/tmp/kvserver"))
        .fn_before_send(|res| match res.message.as_ref() {
            "" => res.message = "altered. Original message is empty.".into(),
            s => res.message = format!("altered: {}", s),
        })
        .into();
    let addr = "127.0.0.1:9527";
    let listener = TcpListener::bind(addr).await?;
    info!("Start listening on {}", addr);
    loop {
        let (stream, addr) = listener.accept().await?;
        info!("Client {:?} connected", addr);
        let svc = service.clone();
        tokio::spawn(async move {
            let mut stream =
                AsyncProstStream::<_, CommandRequest, CommandResponse, _>::from(stream).for_async();
            while let Some(Ok(cmd)) = stream.next().await {
                info!("Got a new command: {:?}", cmd);
                let res = svc.execute(cmd);
                stream.send(res).await.unwrap();
            }
            info!("Client {:?} disconnected", addr);
        });
    }
}

总结：如何用 Rust 做基于 TCP 的网络开发

1. 通过 TcpListener 监听，使用 TcpStream 连接。
2. 在 *nix 操作系统层面，一个 TcpStream 背后就是一个文件描述符。
3. 值得注意的是，当我们在处理网络应用的时候，