一、背景
最近在看Rust
相关的东西,想找个项目用Rust
练下手,熟悉一下Rust
基本特性。然后聊天工具是我们日常最常用的一个软件,我一直想自己写个安全的聊天软件(程序员一般都不相信非开源的程序)。
最终实现的效果图如下(项目地址):
二、技术选型
说到IM
软件,我们常常就会想到一些特性,比如实时性
、安全性
、可靠性
、跨平台兼容性
、消息有序
等等,我们看下常见的一些IM
的技术方案有哪些。
2.1 HTTP 轮询
Http 轮询
顾名思义,通过不停轮询的方式来判断是否有收到新的消息。轮询还分为长轮询
和短轮询
两种。
- 短轮询(
Short Polling
): 短轮询是客户端定期向服务器发送请求,查询是否有新数据。通常,客户端会在每个请求之间设置一个固定时间间隔。以下是短轮询的基本工作流程:- 客户端向服务器发送
HTTP
请求。 - 服务器检查是否有新数据。
- 如果服务器有新数据,立即将数据作为
HTTP
响应返回;如果没有新数据,则直接返回一个空响应或预定义的响应。 - 客户端等待预定的时间间隔,然后再次向服务器发送
HTTP
请求(返回步骤1
)。
- 客户端向服务器发送
- 长轮询(
Long Polling
): 长轮询是短轮询的改进,可以减少服务器负载和网络流量。在长轮询中,客户端发送请求后,服务器会将连接保持打开,直到有新数据可用。以下是长轮询的基本工作流程:- 客户端向服务器发送
HTTP
请求。 - 服务器检查是否有新数据。
- 如果服务器有新数据,立即将数据作为
HTTP
响应返回;如果没有新数据,服务器保持请求打开(一直Hold住),并等待,直到有新数据可用。一旦有新数据,服务器将数据作为HTTP
响应返回。 - 客户端收到响应后,立即向服务器发送新的
HTTP
请求(返回步骤1
)。
- 客户端向服务器发送
优点
- 简单易实现,相较于
TCP Server
,服务端不需要拆包,不需要关注粘包问题,读取Request
、回复Response
都很简单。 - 无需保持长链状态,轮询机制允许客户端和服务器之间在没有实时通信需求时断开连接,服务端也不需要维护链接状态,不需要
Ping
、Pong
保持链接。
缺点
- 延迟高,接受消息的延迟,取决于轮询的间隔。太慢了,延迟大。太快了,无效调用很多。
- 资源浪费,不管是否有消息,客户端都会不停请求服务端,对服务端资源是一种浪费。
这个方案看上去很low
,早些年还是有公司用这种方案快速搭建自己的IM
软件。
2.2 HTTP/2 双向 Stream
HTTP/2
引入了二进制分帧传输、多路复用、首部压缩、服务器推送、请求优先级等特性。HTTP2
的数据传输方式如下图:
HTTP/2
的帧格式如下:
+-----------------------------------------------+
| Length (24) |
+---------------+---------------+---------------+
| Type (8) | Flags (8) |
+-+-------------+---------------+-------------------------------+
|R| Stream Identifier (31) |
+=+=============================================================+
| Frame Payload (0...) ...
+---------------------------------------------------------------+
可以看下Go
对HTTP/2
解码方式如下:
func http2readFrameHeader(buf []byte, r io.Reader) (http2FrameHeader, error) {
_, err := io.ReadFull(r, buf[:http2frameHeaderLen])
if err != nil {
return http2FrameHeader{}, err
}
return http2FrameHeader{
Length: (uint32(buf[0])<<16 | uint32(buf[1])<<8 | uint32(buf[2])),
Type: http2FrameType(buf[3]),
Flags: http2Flags(buf[4]),
StreamID: binary.BigEndian.Uint32(buf[5:]) & (1<<31 - 1),
valid: true,
}, nil
}
由于HTTP/2
支持了二进制分帧
、多路复用
,理论上我们可以直接基于HTTP/2
的双向流(Bidirectional Stream
)的特性来保证消息的实时性,通过HTTPS
来保证消息安全性,还可以通过流量控制
、请求优先级
、来做一些其他更高级的的玩法。
然后我就去找了下有没有HTTP/2
的项目的IM
的库,然后我就找到了一个h2conn 。
// Connect establishes a full duplex communication with an HTTP2 server with custom client.
func (c *Client) Connect(ctx context.Context, urlStr string) (*Conn, *http.Response, error) {
reader, writer := io.Pipe()
// "net/http"
// 这里传 reader 进去,通过writer写入,可以给服务端发送数据
req, err := http.NewRequest(c.Method, urlStr, reader)
if err != nil {
return nil, nil, err
}
......省略部分代码
// Perform the request
resp, err := httpClient.Do(req)
if err != nil {
return nil, nil, err
}
// resp.Body 就是 Reader, 可以读到服务端发过来的数据。
// writer 可以给服务端发送数据
conn, ctx := newConn(req.Context(), resp.Body, writer)
......
return conn, resp, nil
}
服务端要做的就是,收到这个Request
以后,我们handle
函数,一直不返回。不停去读数据和写数据就行了,服务端测试代码如下:
func (h handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
conn, err := h2conn.Accept(w, r)
......
defer conn.Close()
for {
var msg string
err = in.Decode(&msg)
.....
log.Printf("Got: %q", msg)
err = out.Encode(msg)
.....
log.Printf("Sent: %q", msg)
}
}
值得一提的是,这个例子里面,客户端和服务端两边传输的是JSON
的数据流,数据传输的拆包工作是给json.Decoder
去做了。具体代码如下:
// Create a json encoder and decoder to send json messages over the connection
var (
in = json.NewDecoder(conn)
out = json.NewEncoder(conn)
)
// in the decode or encode stages.
for {
var msg string
// {"msg": "a"}{"msg":"b"}
// 这里读到 JSON 字符串流的会自动返回一个完整的 msg
err = in.Decode(&msg)
if err != nil {
log.Printf("Failed decoding request: %v", err)
return
}
log.Printf("Got: %q", msg)
.......
}
虽然说是多路复用,其实本质还是发起一个请求,然后服务端一直Hold
这个请求,不去Close
,然后基于HTTP2
双向流的能力,来收发数据。
这个实现有个缺点就是,如果有一个很大的Request
,会阻塞其他的Request
发送。并不能把这个Request
切分成小块然后分别去发送。
当然也可以专门起一个请求,来接受服务端的推送。但是如果是这样的话,本质就是长轮询
,用HTTP/1
也可以。
2.3 gRPC 双向 Stream
具体实现代码如下:
// proto
service XXX {
rpc StreamTest(stream StreamTestReq) returns (stream StreamTestResp);
}
message StreamTestReq {
int64 i = 1;
}
message StreamTestResp {
int64 j = 1;
}
// server端代码
func (s *XXXService) StreamTest(re v1pb.XXX_StreamTestServer ) (err error) {
for {
data, err := re.Recv()
if err != nil {
break
}
// 将客户端发送来的值乘以10再返回给它
err = re.Send(&v1pb.StreamTestResp{J: data.I * 10 })
}
return
}
// client 端代码
func TestStream(t *testing.T) {
c, _ := service2.daClient.StreamTest(context.TODO())
go func(){
for {
rec, err := c.Recv()
if err != nil {
break
}
fmt.Printf("resp: %v\n", rec.J)
}
}()
for _, x := range []int64{1,2,3,4,5,6,7,8,9}{
_ = c.Send(&dav1.StreamTestReq{I: x})
time.Sleep(100*time.Millisecond)
}
_ = c.CloseSend()
}
// client端输出结果
resp: 10
resp: 20
resp: 30
resp: 40
resp: 50
resp: 60
resp: 70
resp: 80
resp: 90
优点
gRPC
本质底层还是基于HTTP/2
传输,HTTP/2
的优点,gRPC
基本都有。- 强类型和一致的接口:
gRPC
定义了通信协议的标准化、一致且强类型接口。这有助于更好地组织代码,减少错误,并提高可读性和可维护性。 - 有工具可以快速生成客户端和服务端脚手架代码。
缺点
- 浏览器支持不好。
- 单个请求处理所有客户端服务端的通讯数据太重了。
2.4 XMPP
XMPP
(Extensible Messaging and Presence Protocol
,可扩展消息处理协议)和Jabber
实际上是同一个概念的两个不同名称。技术上,它们都指的是同一种基于 XML 的实时通信协议。然而,它们的名称在历史上用于强调不同的方面:
Jabber
Jabber
这个名字来源于最早实现的开放源代码项目,这个项目在1999
年由Jeremie Miller
创建。这个项目旨在实现一种基于互联网且分布式的即时通信协议。当Jabber
项目产生以后,Jabber
这个名称逐渐成为广义上与该项目相关的实时通信协议、技术和工具群的代名词。
XMPP
为了将Jabber
协议变得更加正式和标准化,项目的负责人在2002
年将Jabber
协议提交给了互联网工程任务组(IETF
)。IETF
将Jabber
协议进一步扩展和完善,并最终将其命名为XMPP
(可扩展消息处理协议)。自2004
年起,XMPP
成为了IETF
正式的通信协议标准。
XMPP IM
简单说就是用TCP
传输XML
流,定义了一个数据传输协议的标准,现在基本没有哪家做IM
的公司会遵守这个,所以基本没人用XMPP
了。找了一个开源的Go
的XMMP
的jackal-xmpp。
服务端读取数据的测试代码如下:
// "github.com/jackal-xmpp/stravaganza/parser"
func handleClient(conn net.Conn) {
iq, err := stravaganza.NewBuilder("iq").
WithValidateJIDs(true).
WithAttribute("id", "zid615d9").
WithAttribute("from", "ortuman@jackal.im/yard").
WithAttribute("to", "noelia@jackal.im/balcony").
WithAttribute("type", "get").
WithChild(
stravaganza.NewBuilder("ping").
WithAttribute("xmlns", "urn:xmpp:ping").
Build(),
).
BuildIQ()
if err != nil {
_, _ = fmt.Fprint(os.Stderr, err.Error())
return
}
_ = iq.ToXML(conn, true) // 发送给客户端。
p := xmppparser.New(conn, xmppparser.SocketStream, 1024)
for {
elem, err := p.Parse()
if err != nil {
contine
}
fmt.Println("elem = ", elem.String())
}
}
2.5 自定义协议
早期写TCP Server
的时候,很多时候都是自己去定义数据格式,然后自己读取数据流,然后拆包。
比如斗鱼的弹幕协议:
数据部分用的是自定义的STT
序列化和反序列化。
- 键
key
和值value
直接采用@=
分割 - 数组采用
/
分割 - 如果
key
或者value
中含有字符/
,则使用@S
转义 - 如果
key
或者value
中含有字符@
,使用@A
转义
举例:
- 多个键值对数据:
key1@=value1/key2@=value2/key3@=value3/
value1/value2/value3/
比如登录请求type@=loginreq/roomid@=58839/
最早的是时候还做过类似 PPP(Point-to-Point Protocol) 帧格式,具体编码信息如下:
2.6 QUIC
QUIC(Quick UDP Internet Connections)
在2012
年由Google
开发,它基于UDP(User Datagram Protocol)
而不是TCP
,使用多路复用、内置拥塞控制、低延迟连接建立等技术,为应用提供更快、安全、稳定的端到端传输。
HTTP/2
的一些特性,QUIC
基本上都有。
优点:
- 0-RTT 连接建立,
0-RTT(Zero Round Trip Time)
连接建立是一种特殊的握手机制,允许客户端在重新连接至服务器时(例如在前一次连接中已经与服务器成功建立过TLS
握手的情况下),不需要进行额外的往返时间即可建立新的安全连接。 Quic 加密握手的过程 - 避免线头阻塞:与
HTTP/2
多路复用不同,QUIC
的流在独立的连接中(UDP包),因此可以避免线头阻塞问题。与TCP
传输中存在的因串行处理和线头阻塞导致的潜在延时相比,QUIC
具有更好的吞吐量和时延性能。 - 切换网络时的连接保持,跟上面
0-RTT
一个意思,因为记录了链接密钥和状态(Session Ticket
),所以网络切换(IP
变更)的时候,不需要重新建连。
缺点
- 兼容性:虽然
QUIC
在现代网络环境和浏览器中逐渐被接受,但它的部署和支持相较于HTTP/2
仍有局限性。由于QUIC
依赖于UDP
,一些代理设备、防火墙或者ISP
可能限制了UDP
传输。这可能导致部署和配置的挑战。
QUIC
的帧格式如下:
我说下我使用QUIC
的时候,我的感受。
不需要手工维护 Ping-Pong,
QUIC
自己做了保活机制,我们只用配置一个Keep-Alive
的时间间隔就好了。不需要对数据流做拆包操作,发送数据的流程跟
Http
请求类似,每个Request
和Response
属于不同的StreamID
,在QUIC
中已经帮我们把拆包合包的逻辑做了。我们直接把字节流读出来,转换为Request
或者Response
就行了。
避免线头阻塞,由于
QUIC
是基于UDP
传输的,所以没有TCP
单管道导致的线头阻塞
的问题,如下图所示:没有 UDP 的大包问题,
UDP
长度只有16
位,一般如果超过了65535
,就会被丢弃或者截断。QUIC
会自动帮我去做包拆分,我们写业务层代码的时候,发送数据和使用TCP
写数据流一样,不用Care
发送的数据大小,QUIC
内部会去帮我们做拆包合包。UDP
包结构如下:+---------------------+----------------------+----------------------+ | Source Port (16) | Destination Port (16) | Length (16) | +---------------------+----------------------+----------------------+ | Checksum (16) | Data | +---------------------+-----------------------------------------------+
QUIC
已经是HTTP/3
的标准了,总体来说QUIC
做IM
的传输是一个很好的选择。所以最后准备用QUIC
来做数据传输协议。
三、RUST实现
3.1 异步运行时库 Tokio
Tokio
是Rust
编程语言中的一个异步运行时库,它充分利用Rust
的安全性和并发特性,为开发人员提供高性能、可伸缩的异步I/O
和事件驱动编程能力。简单来说,Tokio
是一个框架,用于通过异步编程构建高效、可伸缩的网络应用程序。
Tokio
的主要特性和组件:
- 运行时:
Tokio
提供了一个运行时,它处理任务调度、执行异步任务和管理操作系统资源。它包含一个高效的事件循环,可以根据输入事件(如:网络数据到达、计时器触发等)执行相应的任务。 - 定时器、任务和工作窃取:
Tokio
提供了定时器功能,允许在指定时间后执行任务;它还包含任务组件,用于创建、管理和在运行时中调度异步任务。为提高性能,Tokio
采用了一个称为“工作窃取”的高效调度策略。 - I/O:
Tokio
库包含对异步I/O
(如TCP
,UDP
, 文件I/O
等)的支持,使您能够轻松构建高性能服务器和客户端。 - 中间件和底层库支持:
Tokio
还配备了一系列中间件和底层库支持,它们提供了常见网络编程任务的便利功能。例如:HTTP
、WebSocket
、加密等。
PS:是不是跟Go
的Runtime
功能有一点点相似。
tokio 应该是RUST
最流行的异步运行时库,看的很多RUST
教程都是直接用的Tokio
。所以我基本上也没调研其他运行时库,选择直接用了Tokio
。
3.2 RUST 的 QUIC 库选择
找了RUST
相关的一些QUIC
库 :
- mozilla/neqo,
mozilla
搞的QUIC
库,依赖NSS/NSPR
两个网络安全库,没有下载NSS/NSPR
直接编译的话,我编译了一个多小时,这个库太重了,直接放弃。 - cloudflare/quiche 一个跨平台的
QUIC
库,很多Low-Level
的API
,扩展性高,但是对新手使用不友好,所以也pass
了。 - aws/s2n-quic,这个是
ASW
的QUIC
库,API
使用也比较简单,不过里面使用了很多宏编程
,对新手来说有一些生涩。 - quinn-rs/quinn,相较于其他几个库,
quinn
的API
简单易用,对新手比较友好。所以,最终使用quinn-rs/quinn这个库。
使用 quinn-rs/quinn API
比较简单,代码如下:
// 发送请求
pub async fn send(&self, request: Request) -> Result<Response> {
let (mut send, mut recv) = self.conn.open_bi().await?;
// 1. read request -> json
let serialized = serde_json::to_vec(&request)?;
// 2. send json data
send.write_all(&serialized).await?;
send.finish().await?;
// 3. read json data
let vec_u8 = recv.read_to_end(MAX_SIZE).await?;
// 4. json data -> Response
let resp: Response = serde_json::from_slice(&vec_u8)?;
Ok(resp)
}
// 接受请求
pub async fn accept_request_loop<F>(&self, callback: F) -> Result<()>
where
F: Fn(Request) -> Result<Response> + Send + Sync + Clone + 'static
{
loop {
let stream = self.conn.accept_bi().await;
let stream = match stream {
Err(e) => {
return Err(SError::Error(format!("Accept stream ,err = {}", e.to_string())));
}
Ok(s) => s,
};
let callback_copy = callback.clone();
let res = handle_request(stream.0, stream.1, callback_copy);
tokio::spawn(async move {
if let Err(e) = res.await {
println!("handle_request failed = {}", e);
}
});
}
}
async fn handle_request<F>(mut send: quinn::SendStream, mut recv: quinn::RecvStream, callback: F) -> Result<()>
where
F: Fn(Request) -> Result<Response> + Send + Sync + Clone
{
// 1. read request json data
let vec_u8 = recv.read_to_end(MAX_SIZE).await?;
// 2. json data -> Request
let request: Request = serde_json::from_slice(&vec_u8).unwrap();
// 3. get response
let resp = callback(request)?;
// 4. response -> json data
let serialized = serde_json::to_vec(&resp)?;
// 5. send response json data
send.write_all(&serialized).await?;
send.finish().await?;
Ok(())
}
3.3 客户端架构选择
早期准备本地启动一个HTTP Server
,然后做一个简单的Web
界面通过HTTP
接口来收发数据。大致流程如下:
后面去调研了下Rust
命令行的UI
库,找到了一个找到了一个还不错的UI
库 fdehau/tui-rs,所以最终决定做成一个命令行的UI
程序。
客户端代码组织的结构,使用了之前做IOS
的MVVM(Model-View-ViewModel)
架构。MVVM
架构是一种针对用户界面(UI
)开发的设计模式。MVVM
旨在将程序中的各个部分分开,以实现关注点分离,从而提高代码的可维护性、可读性和可测试性。MVVM
架构主要由三个核心组件组成:Model
(模型)、View
(视图)和ViewModel
(视图模型)。
Model
(模型):模型表示应用程序的数据和业务逻辑。它暴露了获取数据和执行操作所需的属性和方法。模型是程序的核心,负责实现关键功能和存储业务数据。模型通常与后端服务(如:数据库、APIs
等)进行交互,以读取和持久化数据。View
(视图):视图表示用户界面(UI
),是用户与应用程序进行交互的部分。视图包括屏幕上的所有可见元素和用户可以与之交互的组件(如:文本框、按钮、列表等)。视图一般不包含业务逻辑,它主要负责绑定ViewModel
暴露出的数据和命令,以展示数据和提供交互。ViewModel
(视图模型):视图模型作为模型和视图之间的桥梁,负责在它们之间进行数据转换和通信。视图模型提供了公开要显示在视图上的数据的属性,以及负责处理用户交互的命令。这些命令可以触发业务逻辑的执行(通常在模型层)并更新视图上的数据。视图绑定到视图模型,当视图模型的数据发生变更时,视图会自动更新以反映这些变更。
3.4 消息安全性
在一个消息的收发流程中,通讯链路是是通过TLS
加密过的,理论上是安全的,但是我们的消息是存储在服务端的(即使服务端是加密存储的,服务端自己也可以解密),服务端其实是可以随意查看客户端消息内容的。流程如下图:
3.4.1 方案一:加解密都放在客户端
- 所有客户端上传自己的公钥到服务端。
Client1
要给Client2
、Client3
发送消息的话,需要先请求服务端,拿到Client2
、Client3
的公钥
。Client1
分别使用Client2
和Client3
的公钥对要发送的内容加密,然后把数据发送给服务端。服务端分别把对应的数据转发给对应的客户端。Client2
和Client3
用自己的私钥对收到的密文解密,得到对应的明文。
在这个方案中,我们加密解密都是在客户端做的,所以相较于之前服务度做加密解密会更安全一些。但是服务端这个时候还是可以通过中间人攻击
的方式来Hack
相关的消息内容。具体流程如下:
Client1
请求Client2
的公钥,但是服务端返回的是一个自己的公钥。Client1
在不知情的情况下,使用了服务端的公钥加密,然后把密文发送给服务端。- 服务端这个时候可以拿自己的私钥解密对应的密文,得到明文。
- 服务端使用
Client2
的公钥,对明文进行加密,然后发送给Client2
。 Client2
收到密文以后,正常使用自己的私钥解密。
整个功能对Client
来说都是正常的,但是服务端已经拿到了消息的明文。
3.4.2 方案二:P2P协商公钥
这个方案一的本质问题还是因为依赖了服务端去交换各种的公钥,所以服务端可以作为中间人用假的公钥去替换客户端的公钥。我们可以基于P2P(Peer-to-Peer)
点对点通讯的方式去互相交换公钥就可以了。 具体流程如下:
- 客户端链接服务端。服务端记录所有客户端的
IP
地址。 - 客户端请求服务端,获取当前群中所有
Client
的IP
地址。 - 客户端互相发送
Ping-Pong
的UDP
消息来打通双发链路信息(这一步就是打洞
)。 P2P
的链路通了以后,就可以互相交换公钥了。- 使用交换以后公钥来加密消息,通过服务端转发消息。(走服务端发送消息的好处是,服务端可以存储消息,部分客户端离线以后再上线还能收到之前的消息)
值得一提的是P2P
打洞不是所有网络环境都能成功,跟路由的NAT
类型有关,路由器的NAT
类型可以分为几类:
- 全锥 NAT(
Full Cone NAT
) - 限制性锥 NAT(
Restricted Cone NAT
) - 端口限制性锥 NAT(
Port Restricted Cone NAT
) - 对称 NAT (
Symmetric NAT
)
在对称NAT
的情况下(很少有这种类型的路由器),是会打洞失败的。这种情况应该兜底走方案一。P2P
的穿透/打洞原理,这里就不过多赘述,感兴趣的朋友可以自己去搜索相关文章。
笔者之前做P2P
项目的时候,分别测试过电信和联通的路由器:
- 电信的更多是
全锥NAT
型路由器,只要知道了客户端的IP:Prot
, 任何一个外部主机均可通过该IP:Port
发送数据包到该主机。 - 联通的更多是
端口限制性锥NAT
:即只有内部主机先向外部IP:Prot
发送数据包,该外部主机才能使用特定的端口号向内部主机发送数据包。
四、总结
做了个简单聊天室功能, 熟悉了一下Rust
一些的基本特性。 Always Exploring…