前言#
有好一段时间没碰rust了，有些痒。。
这下面涉及到的项目，我放到git上去了，不想敲代码可以直接拉。
https://github.com/1714080902120/rust_mini_redis_demo.gittokio#
tokio是rust异步未来的一个方向，由于rust自身异步这块并不完善，所以就目前来说，tokio等异步runtime都有可能被合并进rust里面又或者被官方指定为唯一库等。
我们之前学过rust原生的异步，但是官方的文档并不完善。
感兴趣的可以去看下：坏蛋Dan：rust基础学习索引--异步
话不多说，咱直接开始学习
PS: 下面提到的原生都指的是futures[2]这个crate。
简介#
简单地说就是一个异步rust runtime。
它主要有以下几个组件：
- 一个多线程
runtime用于执行异步代码 - 一个标准库的异步版本
- 一个生态丰富的库
中间有一大段自夸就不介绍了，简单的说就是：快、稳、简单、灵活。
什么情况下不能使用tokio#
- 多线程并行(
parallel)高速CPU密集型计算(Speeding up CPU-bound computations) 。tokio是设计用于IO密集型应用的，也就是说它里面的任务会花更多的时间等待IO。如果你就是想要并行密集型计算，麻烦出门左拐:rayon[3]。 - 大批量文件读写。虽然看起来非常适合，但实际开发中
tokio在这块并不会比线程池强。因为一般操作系统是不会提供异步文件的API的。 - 单个
web请求，tokio是用来处理同时多个任务的。当然，也不是不能用，不过和使用block api的对比没什么优势。
请求帮助#
如果遇到了什么问题，可以直接去以下两个地方请求帮助：
尤其是github，一般对于新的库或者新的语言来说，搜索引擎的作用就不太大了。
不过好处是你可以和库/语言作者深入(deep)♂的沟通~
基础api和用法#
spawning#
需要注意的是这里的spawning和std::thread::spawn不是同一个东西，后者表示创建一个新的线程，而前者表示创建一个新的异步任务(task)。
我们来看下基础用法
use tokio::net::TcpListener;
#[tokio::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
loop {
let (socket, _) = listener.accept().await.unwrap();
// A new task is spawned for each inbound socket. The socket is
// moved to the new task and processed there.
tokio::spawn(async move {
process(socket).await;
});
}
}这里的TcpListener是tokio自己实现的，和其它第三方异步库比如async-std一样，名字都是和标准库里的一样，但是是异步的。
这里有一点需要注意，spawn里的block需要接收数据的所有权，因为tokio虽然是处理异步，但是底层是任务调度是多线程的。
不用担心这里存在严重损耗，实际上它是轻量的，仅占用内存64个字节的空间。
spawn接收一个Future,返回一个joinHandle。
同时数据也需要实现Send和'static，其中的原因都是和上面需要数据的所有权一样，都是因为tokio底层任务调度是多线程的。
这里说一下这个'static，我们第一次接触到这个的时候是悬浮引用那一章，比如下面这样
fn hangReference () -> &str {
let a = "123";
a
}这段代码是会报错的，因为a在这个函数中执行完出了调用栈之后就会被drop了，所以如果此时外部还有指针，那么指向的地址就可能是错的了，即使是这个"123"会被硬编码到代码中，与整个runtime长存。
这个时候使用'static通知编译器这个引用是没问题的
fn hangReference () -> &'static str {
let a = "123";
a
}不过一般我们都不这样写，如果你有需要，直接写到全局即可。
回到我们的代码中，这里的'static只是lifetime bound，告知编译器这个异步block里的数据不能有任何外部引用存在，这么做就保证了异步block里的数据一定是这个异步block的。
关于'static，这里有一篇文章是解释一些常见的误解的，比如'static表示这个数据一定与runtime长存这种：https://github.com/pretzelhammer/rust-blog/blob/master/posts/common-rust-lifetime-misconceptions.md#2-if-t-static-then-t-must-be-valid-for-the-entire-program
其它也就不过多介绍了。
如果你的异步block有返回数据，那么你可以对spawn返回的joinHandle做await，拿到的是一个Result包裹的数据。
比如：
#[tokio::main]
async fn main() {
let handle = tokio::spawn(async {
// Do some async work
"return value"
});
// Do some other work
let out = handle.await.unwrap();
println!("GOT {}", out);
}
channels#
tokio提供了以下四种不同类型的channels：
mpsc[6]：多个producer，单个consumer，适用于接收多个值oneshot[7]：一个producer和一个consumer，适用于单个值的场景broadcast[8]：多个producer和多个consumer，适用于很多值的场景，每个receiver都能接触到每个值。watch[9]：单个producer，多个consumer。适用于很多值的场景，但是receiver只会接触到最近的一个值。
如果你需要的是多个producer和多个consumer但是只需要其中一个receiver接触到所有值的场景，那么你可以使用async-channel[10]这个crate。
当然，前提都是异步的场景，如果你的代码不需要异步，那么直接用标准库的std::sync::mpsc[11]即可，后者第三方crate：crossbeam::channel[12]。
我们接下来只了解mpsc和oneshot两种，另外的如果你感兴趣可自行了解。
我们先来看下tokio::sync::mpsc的用法
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
let tx2 = tx.clone();
tokio::spawn(async move {
tx.send("sending from first handle").await;
});
tokio::spawn(async move {
tx2.send("sending from second handle").await;
});
while let Some(message) = rx.recv().await {
println!("GOT = {}", message);
}
}
和std::sync::mpsc用法基本一模一样。我们创建的channel容量是32，当message太多,达到了receiver可承受的上限，那么这个时候多出来的任务就会被存储到容器里sleep，等到可以处理该信息之后才会从容器中移除。
我们可以通过clone方法来将发射器传递给不同的任务，但是接收器就不行了，对于mpsc来说，一个channel只能有一个接收器。
当没有发射器可以发送数据之后，接收器recv拿到的值则是None，表示这个channel已经关闭了。
然后我们再来看下oneshot的基础用法
use tokio::sync::oneshot;
#[tokio::main]
async fn main() {
tokio::spawn(async move {
let (tx, rx) = oneshot::channel();
tx.send("test").unwrap();
let res = rx.await.unwrap();
dbg!(res);
});
}
和mpsc差不多，不过这里接收器和发射器只允许有一个，即都不能复制。使用send的时候并不需要.await
这里也并不需要传入容器的大小，因为是一对一的。
I/O#
tokio里的异步I/O用法上和标准库里的差不多，不过是异步的，每一步使用都需要.await。
读是基于AsyncRead，写是基于AsyncWrite。TcpStream、File以及Stdout这些trait都是实现了这俩trait的。
另外AsyncWrite还被Vec<u8>以及&[u8]实现了，因为大部分场景都是处理bytes，这样我们用起来就比比较方便。
另外我们基本并不直接使用AsyncRead和AsyncWrite这俩，就像我们不直接使用Future的poll一样。
我们基本都是通过AsyncReadExt和AsyncWrite这俩提供的方法来间接使用到AsyncRead和AsyncWrite。
我们来看下基本用法
AsyncReadExt::read：这是个异步方法用于将数据读到一个buffer中，返回读取的字节数量。注意，一旦返回变成Ok(0)，那么就意味着这个读取流已经关闭了。而在TcpStream中则表示这个socket结束了。
use tokio::fs::File;
use tokio::io::{self, AsyncReadExt};
#[tokio::main]
async fn main() -> io::Result<()> {
let mut f = File::open("foo.txt").await?;
let mut buffer = [0; 10];
// read up to 10 bytes
let n = f.read(&mut buffer[..]).await?;
println!("The bytes: {:?}", &buffer[..n]);
Ok(())
}AsyncReadExt::read_to_end：和上面的一样，不过它会一直读直到EOF(end of file)。
use tokio::io::{self, AsyncReadExt};
use tokio::fs::File;
#[tokio::main]
async fn main() -> io::Result<()> {
let mut f = File::open("foo.txt").await?;
let mut buffer = Vec::new();
// read the whole file
f.read_to_end(&mut buffer).await?;
Ok(())
}AsyncWriteExt::write：将buffer写入到writer中，返回写入的bytes长度。
use tokio::io::{self, AsyncWriteExt};
use tokio::fs::File;
#[tokio::main]
async fn main() -> io::Result<()> {
let mut file = File::create("foo.txt").await?;
// Writes some prefix of the byte string, but not necessarily all of it.
let n = file.write(b"some bytes").await?;
println!("Wrote the first {} bytes of 'some bytes'.", n);
Ok(())
}AsyncWriteExt::write_all：不多说。
use tokio::io::{self, AsyncWriteExt};
use tokio::fs::File;
#[tokio::main]
async fn main() -> io::Result<()> {
let mut file = File::create("foo.txt").await?;
file.write_all(b"some bytes").await?;
Ok(())
}另外tokio::io还包含了一些辅助函数，这些函数一般用于标准输入/输出/错误中。
比如tokio::io::copy这个辅助函数可以复制整个reader到writer里面。
use tokio::fs::File;
use tokio::io;
#[tokio::main]
async fn main() -> io::Result<()> {
let mut reader: &[u8] = b"hello";
let mut file = File::create("foo.txt").await?;
io::copy(&mut reader, &mut file).await?;
Ok(())
}注意这里用的reader和writer的bytes都是实现了AsyncRead/AsyncWrite的。
然后我们来搞一个简单的demo来加深下印象。
cargo init echo-server别忘了workspace.members中补充下它的名字
我们的这个server的功能是接收输入的数据，copy一下，然后将数据返回出去，看着有些蛋疼，但是学习嘛。
首先是整体，自然是有个TcpListener，然后loop轮询等待请求，遇到请求则创建一个异步任务来处理
use tokio::io;
use tokio::net::TcpListener;
#[tokio::main]
async fn main() -> io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:6142").await?;
loop {
let (mut socket, _) = listener.accept().await?;
tokio::spawn(async move {
// Copy data here
});
}
}然后我们就遇到了一个问题，我们刚学的io::copy需要两个参数，但我们这只有一个。不过不用担心，socket自身即实现了AsyncRead，又实现了AsyncWrite，不过这里还有个问题，就是copy需要的是&mut，也就是说得同时存在两个&mut。
如果直接写编译器是不给过的，原理相信大家都清楚。根据我们所学知识，我们其实还是有方法可以绕过这种问题的， 不过都是unsafe的，比如原始指针等，但我们并不需要手动去实现，官方已经提供了对应的方案：io::split。
可以看到它里面用了Arc智能指针做了一层封装，其实还是指向同一个内存空间。然后通过lock也就是锁的方式来解决资源竞争的问题。
教程里说这里还用到了mutex，但是我看了下代码并没有（可能有，但并不体现在这里，因为这里还涉及到了几个宏，跟踪有些麻烦），这里实际上开锁解锁用的类似Future内部poll的效果，等到准备好可以解锁了就会自动解锁。
TcpStream也有split方法
简单暴力，好处是没有runtime损耗，坏处自然就是只能在同一个任务里。
use tokio::{
io::{self},
net::TcpListener,
};
#[tokio::main]
async fn main() -> io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:6142").await?;
loop {
let (mut socket, _) = listener.accept().await?;
tokio::spawn(async move {
let (mut rd, mut wt) = socket.split();
let num = io::copy(&mut rd, &mut wt).await.unwrap();
dbg!(num);
});
}
}效果就不用看了，没意义。。
然后我们来实现不使用copy的方式
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
#[tokio::main]
async fn main() -> io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:6142").await?;
loop {
let (mut socket, _) = listener.accept().await?;
tokio::spawn(async move {
let mut buf = vec![0; 1024];
loop {
match socket.read(&mut buf).await {
// Return value of `Ok(0)` signifies that the remote has
// closed
Ok(0) => return,
Ok(n) => {
// Copy the data back to socket
if socket.write_all(&buf[..n]).await.is_err() {
// Unexpected socket error. There isn't much we can
// do here so just stop processing.
return;
}
}
Err(_) => {
// Unexpected socket error. There isn't much we can do
// here so just stop processing.
return;
}
}
}
});
}
} 代码很简单，就是使用了read和write_all两个方法。
这里要注意Ok(0)的时候停止loop，要不然会一直执行下去。
另外有一点需要说下，这里用来存储buf的是vec也就是堆数组，实际上用栈数组也不是不行，但是编译之后会导致页面大小变得非常大，原因是栈数组会inline到页面中，类似一个枚举，所以使用堆数组较好。
Framing#
Framing是指将获取到的字节流(bytes stream)转换成帧流(frame stream)的过程。帧是两个对等体(two peers，也可以理解为channel双方)之间的数据单元。
比如redis中对于帧的定义
use bytes::Bytes;
enum Frame {
Simple(String),
Error(String),
Integer(u64),
Bulk(Bytes),
Null,
Array(Vec<Frame>),
} 又比如http的帧
enum HttpFrame {
RequestHead {
method: Method,
uri: Uri,
version: Version,
headers: HeaderMap,
},
ResponseHead {
status: StatusCode,
version: Version,
headers: HeaderMap,
},
BodyChunk {
chunk: Bytes,
},
} 这个比上面redis的好懂多了。。
select#
我们之前学习异步的时候也学过这货: rust基础学习--异步day6 - 知乎 (zhihu.com)
直接来看个例子
use tokio::sync::oneshot;
#[tokio::main]
async fn main() {
let (tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel();
tokio::spawn(async {
let _ = tx1.send("one");
});
tokio::spawn(async {
let _ = tx2.send("two");
});
tokio::select! {
val = rx1 => {
println!("rx1 completed first with {:?}", val);
}
val = rx2 => {
println!("rx2 completed first with {:?}", val);
}
}
}简单的说就是promise.race[15]，当其中一个ready之后就返回。源码咱就不看了，因为是一个声明宏，可读性极其差。里面用一个大的future来包裹传进去的几个future，里面就是有一个全局状态，如果is_pending一直是true，那么就一直是pending，当这个is_pending变成false之后才会Ready。
而剩下还在pending中的future将会被drop处理。
我们也可以基于这一点来实现任务的cancel，比如：
use tokio::sync::oneshot;
async fn some_operation() -> String {
// Compute value here
"test".to_string()
}
#[tokio::main]
async fn main() {
let (mut tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel();
tokio::spawn(async {
// Select on the operation and the oneshot's
// `closed()` notification.
tokio::select! {
val = some_operation() => {
let _ = tx1.send(val);
}
_ = tx1.closed() => {
// `some_operation()` is canceled, the
// task completes and `tx1` is dropped.
}
}
});
tokio::spawn(async {
let _ = tx2.send("two");
});
tokio::select! {
val = rx1 => {
println!("rx1 completed first with {:?}", val);
}
val = rx2 => {
println!("rx2 completed first with {:?}", val);
}
}
}如果tx1.closed先执行完，然后some_operation这里的future就会被drop，那么tx1.send(val)就不会执行。原理简单的说就是receiver向Sender发送了一个close的提醒，然后Sender``drop了自己。
为了加深下印象，这里用一个例子来模拟底层的原理
use tokio::sync::oneshot;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
struct MySelect {
rx1: oneshot::Receiver<&'static str>,
rx2: oneshot::Receiver<&'static str>,
}
impl Future for MySelect {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
if let Poll::Ready(val) = Pin::new(&mut self.rx1).poll(cx) {
println!("rx1 completed first with {:?}", val);
return Poll::Ready(());
}
if let Poll::Ready(val) = Pin::new(&mut self.rx2).poll(cx) {
println!("rx2 completed first with {:?}", val);
return Poll::Ready(());
}
Poll::Pending
}
}
#[tokio::main]
async fn main() {
let (tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel();
// use tx1 and tx2
MySelect {
rx1,
rx2,
}.await;
}
MySelect`里面就是我们传过去的`future`，然后我们给`MySelect`实现`Futuretype Output是关联类型，用于future的返回，我们这里没有返回值，所以设置为空元组。
future这个trait的poll方法是必须实现的，一般我们都不会去手动实现它，它自己内部会调用这个poll。这个方法里就是你如何确定ready的逻辑，我们这里是等待两个future其中一个ready之后就返回，否则一直是pending。
正如我前面说的，他就是用一个future来包裹传进去的future，然后等待其中一个ready了就直接ready并且return。
select!`的语法是`<pattern> <async expression> <handler>`，不过需要注意的是最多只支持`64`条`arm
最先到达终点的async expression会返回一个值给pattern，然后这个pattern可以被handler使用。
不过既然是pattern，那么作用就很多了。当你跑的最快的那个future返回的值和pattern的不匹配，那么它就会等待下一个future，直到遇到了匹配上的pattern才会结束.
比如：
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (mut tx1, mut rx1) = mpsc::channel(128);
let (mut tx2, mut rx2) = mpsc::channel(128);
tokio::spawn(async move {
// Do something w/ `tx1` and `tx2`
});
tokio::select! {
Some(v) = rx1.recv() => {
println!("Got {:?} from rx1", v);
}
Some(v) = rx2.recv() => {
println!("Got {:?} from rx2", v);
}
else => {
println!("Both channels closed");
}
}
}只有成功recv的时候才会匹配到，如果遇到None则不匹配，于是又会等待下一条armready
这里还有一个特殊用法，就是else，用于兜底。
这里还有一点需要说下：所有权问题。
前面我们使用tokio::spawn的时候如果引用到外部的数据，那么就一定需要获取所有权，原因是tokio底层是多线程异步，所以数据在传递的过程必须要有所有权并且是Send的。
而select!并不需要，因为select!中不可能会有两条arm同时实现并且匹配最终handle，如果失败了就drop，成功了则drop剩下的future。所以不存在数据竞争的问题。比如：
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use std::io;
use std::net::SocketAddr;
async fn race(
data: &[u8],
addr1: SocketAddr,
addr2: SocketAddr
) -> io::Result<()> {
tokio::select! {
Ok(_) = async {
let mut socket = TcpStream::connect(addr1).await?;
socket.write_all(data).await?;
Ok::<_, io::Error>(())
} => {}
Ok(_) = async {
let mut socket = TcpStream::connect(addr2).await?;
socket.write_all(data).await?;
Ok::<_, io::Error>(())
} => {}
else => {}
};
Ok(())
}select!一般较多用于loop的场景，比如下面这个例子：
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx1, mut rx1) = mpsc::channel(128);
let (tx2, mut rx2) = mpsc::channel(128);
let (tx3, mut rx3) = mpsc::channel(128);
loop {
let msg = tokio::select! {
Some(msg) = rx1.recv() => msg,
Some(msg) = rx2.recv() => msg,
Some(msg) = rx3.recv() => msg,
else => { break }
};
println!("Got {:?}", msg);
}
println!("All channels have been closed.");
}比如一个聊天室，所有人的信息都需要展示在聊天框里，顺序上自然是哪个快上哪个，循环等待有人发信息直到所有人都返回None表示close。
前面我们用的都是channel的，如果我们只是写一个普通的异步函数，那么这个时候就需要先用到tokio::pin!来处理下future，不然会有问题。
我们以前分析过pin这个东西，这里就不多说了，简单地说就是让它的指针指向的数据不会被移动走：具体分析可以看rust基础学习--异步day4 - 知乎 (zhihu.com)
把operation放在loop外面，这样每次迭代都指向同一个数据
我们再来看个例子，等待分支中拿到的值是奇数才停止循环。
async fn action(input: Option<i32>) -> Option<String> {
// If the input is `None`, return `None`.
// This could also be written as `let i = input?;`
let i = match input {
Some(input) => input,
None => return None,
};
// async logic here
}
#[tokio::main]
async fn main() {
let (mut tx, mut rx) = tokio::sync::mpsc::channel(128);
let mut done = false;
let operation = action(None);
tokio::pin!(operation);
tokio::spawn(async move {
let _ = tx.send(1).await;
let _ = tx.send(3).await;
let _ = tx.send(2).await;
});
loop {
tokio::select! {
res = &mut operation, if !done => {
done = true;
if let Some(v) = res {
println!("GOT = {}", v);
return;
}
}
Some(v) = rx.recv() => {
if v % 2 == 0 {
// `.set` is a method on `Pin`.
operation.set(action(Some(v)));
done = false;
}
}
}
}
}这里唯一不了解的地方应该是operation.set(action(Some(v)))这个。实际上.set是来自于Pin的方法，看下源码
很好理解，就是修改指向地址的数据.
所以这里其实就是在给operation修改值，然后.await的时候返回给res。
streams#
stream指的是一些列异步的值，你可以当作是异步的迭代器，也就是std::iter::Iterator。
目前tokio用另一个crate tokio-stream来提供对stream的支持，一旦原生的Stream稳定下来后这个crate就会被整合到tokio里面。
虽然可以视作是迭代器，但是它并不能用for来遍历，不过我们还是可以用while let来搭配StreamExt::next()迭代的。比如：
use tokio_stream::StreamExt;
#[tokio::main]
async fn main() {
let mut stream = tokio_stream::iter(&[1, 2, 3]);
while let Some(v) = stream.next().await {
println!("GOT = {:?}", v);
}
}这里还有个稍微复杂点的，基于mini_redis::client提供的发布订阅。
use tokio_stream::StreamExt;
use mini_redis::client;
async fn publish() -> mini_redis::Result<()> {
let mut client = client::connect("127.0.0.1:6379").await?;
// Publish some data
client.publish("numbers", "1".into()).await?;
client.publish("numbers", "two".into()).await?;
client.publish("numbers", "3".into()).await?;
client.publish("numbers", "four".into()).await?;
client.publish("numbers", "five".into()).await?;
client.publish("numbers", "6".into()).await?;
Ok(())
}
async fn subscribe() -> mini_redis::Result<()> {
let client = client::connect("127.0.0.1:6379").await?;
let subscriber = client.subscribe(vec!["numbers".to_string()]).await?;
let messages = subscriber.into_stream();
tokio::pin!(messages);
while let Some(msg) = messages.next().await {
println!("got = {:?}", msg);
}
Ok(())
}
#[tokio::main]
async fn main() -> mini_redis::Result<()> {
tokio::spawn(async {
publish().await
});
subscribe().await?;
println!("DONE");
Ok(())
}
同样需要注意使用tokio::pin!来包裹异步行为，避免迭代中改变了数据的位置。
然后我们来了解下stream的适配器(adapters)，简单地说就是提供操作stream的方法，比如map、take、filter等。
先来看下take的作用：
// ...
let messages = subscriber
.into_stream()
.take(3);
// ...还是刚mini_redis的例子，不过接收数据的时候take了下，然后只会接收前三个。
然后是filter:
// ...
let messages = subscriber
.into_stream()
.filter(|msg| match msg {
Ok(msg) if msg.content.len() == 1 => true,
_ => false,
})
.take(3);
// ...
选取其中文本长度为1的前三个
最后来看下map：
// ...
let messages = subscriber
.into_stream()
.filter(|msg| match msg {
Ok(msg) if msg.content.len() == 1 => true,
_ => false,
})
.map(|msg| msg.unwrap().content)
.take(3);
// ...
实战#
直接实战上手，狠库话不多。
我们来一步一步构建一个Redis[18]客户端和服务器。至于Redis是啥相信不用我多说大家都清楚。
这里我们将实现一个简单的Redis命令的子集，然后在这个过程中学习Tokio。
项目结构#
我们的项目分为两个部分，一部分是server，另一部分是client。
其中server的作用是接收并存储数据或者返回数据给client。
而client则是提供相关api给用户，接收用户传入的值并发送给server。
Mini-Redis server#
我们先安装下mini-redis这个包，我们用它先来写一个demo效果。
cargo install mini-redis安装完之后测试下正不正常
先开一个终端跑起来服务器
mini-redis-server然后另外开个终端
mini-redis-cli get foo展示nil则为正常
初始化项目#
我这里就直接搞个workspaces了，如果不想搞可以不配置最外层的Cargo.toml
mkdir tokio_study
cd tokio_study
echo > Cargo.toml
cargo new my-redis
cd my-redis然后配置下最外层的Cargo.toml
[workspace]
members = ["my-redis"]将我们的client添加到workspace之后我们再来配置下my-redis
[package]
name = "my-redis"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio = { version = "1", features = ["full"] }
mini-redis = "0.4"我们需要tokio和mini-redis。
full表示全量引入，有的时候我们仅用到部分功能，那么这个时候就可以选择只引入用到的部分。
不过开发阶段没这么多要求，全量即可。
那么就配置好了
demo#
进入到main.rs文件中
use mini_redis::{ client, Result };
#[tokio::main]
async fn main() -> Result<()> {
let mut client = client::connect("127.0.0.1:6379").await?;
client.set("hello", "tokio".into()).await?;
let result = client.get("hello").await?;
println!("get value from server: result={:?}", result);
Ok(())
} #[tokio::main]这个是属性宏，这里就不多说了，简单地说就是编译阶段会触发这个宏然后执行对应的函数处理抽象语法树再转换回代码，比如这里将async转换为原生block_on的写法，毕竟main函数是不允许async的。
// 转换前
#[tokio::main]
async fn main() {
println!("hello");
}
// 转换后
fn main() {
let mut rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
println!("hello");
})
}?操作符简单地说是用于简写错误传递的，可以理解为语法糖，如果不记得了可以回去看下我之前的文章: rust基础学习--day19 - 知乎 (zhihu.com)(其实是我自己不记得了。。)Result<()>：这里的Result实际上就是标准库里的Result，至于为什么只需要传一个类型，自然是做了一层Wrapper
.into是实现了Into这个trait，我们来看下源码简单的理解下
它实际上内部走的是U::from(self)的逻辑，即将自己变成类型U。 T在这自然就是&str这个类型, 而我们这里的类型U则是Bytes这个类型， 所以Bytes源码一定会有实现From这个trait
所以上面的tokio.into()是实际上相当于
Bytes.from("tokio")扯远了。。。。
我们基于mini_redis::client创建了一个客户端，异步connect上了本机之前开启的服务器的6379端口。
既然连接都异步了，那后面的步骤自然就都需要异步处理。
然后set了一个key为hello值为tokio的数据。
然后我们cargo run（或者cargo run -p my-redis） （别忘了先开服务器，mini-redis-server）
正常拿到值，不过注意拿到的是Bytes类型的数据。
现在我们就像是在写同步代码一样，不过这一点很多库里面都一样，甚至是用原生feature的async/await语法糖来写也没什么问题，毕竟代码量不是很大。
服务器：基于HashMap实现读取#
上面的demo代码我们放到别的项目中，比如demo文件夹中。
然后我们准备来实现一个自己的server。
(别忘了加到workspace中)
既然是服务器，那么第一步自然就是跑起来和监听某个接口以及响应请求
use tokio::net::{TcpListener, TcpStream};
use mini_redis::{Connection, Frame};
#[tokio::main]
async fn main() {
// Bind the listener to the address
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
loop {
// The second item contains the IP and port of the new connection.
let (socket, _) = listener.accept().await.unwrap();
process(socket).await;
}
}
async fn process(socket: TcpStream) {
// The `Connection` lets us read/write redis **frames** instead of
// byte streams. The `Connection` type is defined by mini-redis.
let mut connection = Connection::new(socket);
if let Some(frame) = connection.read_frame().await.unwrap() {
println!("GOT: {:?}", frame);
// Respond with an error
let response = Frame::Error("unimplemented".to_string());
connection.write_frame(&response).await.unwrap();
}
}TcpListener：这个是tokio自己实现的async版本的TcpListener，和其它库比如async-std一样都是直接用同样的名字，但是都是异步的。Connection：这个来自于mini_redis这个包里的，也是异步的。它自己封装了一层，这样我们就不用去自己去操作tcpStream的buffer了。read_frame：返回”一帧“, 这个帧是基于tcpStream的，这个是tokio自己设计的一个类型，这个帧这里可以简单的理解为是一次请求的请求体，而它里面实际上是从stream里面拿固定的一部分，下次再拿剩余里的一部分。write_frame：将我们的内容转换成buffer再传输响应。
以上行为都是异步的。
然后我们来运行下，先cargo run -p my-redis将服务跑起来，然后再另起一个终端cargo run -p demo
目前的响应应该如下
而服务器端表现应该如下
目前我们仅可以返回unimplemented这个错误以及无法存取传入的数据，我们来完善这一部分。
use tokio::net::{TcpListener, TcpStream};
use mini_redis::{Connection, Frame};
#[tokio::main]
async fn main() {
// Bind the listener to the address
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
loop {
// The second item contains the IP and port of the new connection.
let (socket, _) = listener.accept().await.unwrap();
process(socket).await;
}
}
async fn process(socket: TcpStream) {
use mini_redis::Command::{self, Get, Set};
use std::collections::HashMap;
// A hashmap is used to store data
let mut db = HashMap::new();
// Connection, provided by `mini-redis`, handles parsing frames from
// the socket
let mut connection = Connection::new(socket);
// Use `read_frame` to receive a command from the connection.
while let Some(frame) = connection.read_frame().await.unwrap() {
let response = match Command::from_frame(frame).unwrap() {
Set(cmd) => {
// The value is stored as `Vec<u8>`
db.insert(cmd.key().to_string(), cmd.value().to_vec());
Frame::Simple("OK".to_string())
}
Get(cmd) => {
if let Some(value) = db.get(cmd.key()) {
// `Frame::Bulk` expects data to be of type `Bytes`. This
// type will be covered later in the tutorial. For now,
// `&Vec<u8>` is converted to `Bytes` using `into()`.
Frame::Bulk(value.clone().into())
} else {
Frame::Null
}
}
cmd => panic!("unimplemented {:?}", cmd),
};
// Write the response to the client
connection.write_frame(&response).await.unwrap();
}
}这里我们基于标准库里集合的HashMap来实现数据的存取。
Command::from_frame：是将原来接收到的数组转换成一个Command枚举的变体。中间处理有些绕，这里就不分析源码了。
Frame::Simple：这个和下面的Frame::Bulk以及Frame::Null都表示的是帧的状态，这些状态根据的是redis的协议
我们现在只处理set和get两种场景，其它先返回错误。set的时候将value从Bytes类型转换成Vector，然后get的时候又转换回来。
我们来重新运行下
这回响应就正常了。
不过现在我们的数据依旧不是持久化的，当我们的服务器关闭后数据就都不见了。接下来我们来实现数据持久化。
服务器：数据共享#
在我们处理数据持久化问题之前，我们需要处理一个问题：数据并不共享，目前我们的数据HashMap并不共享，而是在process方法中给每个连接对象创建一个hashmap作为state也就是数据。
我们应该将它拿出来作为共享数据，不过这里又遇到了一个问题，hashmap的get和insert都不是异步的，这也就意味着可能会有并发问题。
go`语言文档中存在这么一句话： `Do not communicate by sharing memory; instead, share memory by communicating.前面无畏并发那篇文章中我们说到过rust也是倾向于上面这句话的理念：不要通过分享内存来通信，而是通过通信来共享内存。
我们有两种方式来解决并发问题：
- 通过
Mutex来管理数据共享。 - 创建任务管理状态并通过通信的方式去操作数据。
我们先用第一种方案来解决这个问题，之后我们会遇到用第二个方案来处理的。
在支持共享之前，我们先优化下我们存储数据的类型，现在我们每次get都是clone一份Vec<u8>然后转换成Bytes这个类型，然后set的时候都是从Bytes转换成Vec<u8>。这样其实完全没必要，另外一个点就是Bytes类型clone的时候只是浅复制，只会创建一个指针。
可以看到数据类型AtomicPtr实际上是一个UnsafeCell。这个UnsafeCell简单地说就是一个标记类型，它返回一个原始指针，可以让你绕过编译，允许你在runtime直接修改&T指向的数据。 一般用在并发场景，和mutex、refcell等搭配。
扯远了。。。
我们先引入下对应的包，由于孤岛规则，即使原有的包中带有这个Bytes我们也是无法直接使用的
[package]
name = "my-redis"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio = { version = "1", features = ["full"] }
mini-redis = "0.4"
bytes = "1" 然后我们回到main.rs文件中修改下对应的代码，将我们的HashMap变成一个支持多线程读写的数据类型
use bytes::Bytes;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
type Db = Arc<Mutex<HashMap<String, Bytes>>>;Arc和Mutex这俩货相信大家都懂，如果忘记了可以去看之前的文章：
不过这里需要注意的是，我们用的是标准库里的Arc和Metux，而不是tokio里的Arc和Metux，因为HashMap自身读写都是同步的，所以用了tokio里的也没效果。
然后我们来修改相关代码
use mini_redis::{Connection, Frame};
use bytes::Bytes;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
type Db = Arc<Mutex<HashMap<String, Bytes>>>;
use tokio::net::{TcpListener, TcpStream};
#[tokio::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
println!("Listening");
let db = Arc::new(Mutex::new(HashMap::new()));
loop {
let (socket, _) = listener.accept().await.unwrap();
// Clone the handle to the hash map.
let db = db.clone();
println!("Accepted");
tokio::spawn(async move {
process(socket, db).await;
});
}
}
async fn process(socket: TcpStream, db: Db) {
use mini_redis::Command::{self, Get, Set};
// Connection, provided by `mini-redis`, handles parsing frames from
// the socket
let mut connection = Connection::new(socket);
while let Some(frame) = connection.read_frame().await.unwrap() {
let response = match Command::from_frame(frame).unwrap() {
Set(cmd) => {
let mut db = db.lock().unwrap();
db.insert(cmd.key().to_string(), cmd.value().clone());
Frame::Simple("OK".to_string())
}
Get(cmd) => {
let db = db.lock().unwrap();
if let Some(value) = db.get(cmd.key()) {
Frame::Bulk(value.clone())
} else {
Frame::Null
}
}
cmd => panic!("unimplemented {:?}", cmd),
};
// Write the response to the client
connection.write_frame(&response).await.unwrap();
}
}和之前的代码其实差别不大，调整的点在于开了多线程，db是通过参数传入以及读写不再需要转换类型。
现在我们的db是共享的了，并且通过互斥锁解决了并发问题。
不过这里又出现了另一个问题：堵塞。
由于锁仅能有一个使用者，那么这个时候其它线程都需要堵塞等待锁的权限才行，这样就会造成堵塞问题。
即使我们不开多线程也是会存在这个问题的，因为tokio底层就是基于多线程调度器(scheduler)。
那么要如何解决这个问题呢？你可能会想这个时候使用tokio的mutex能否解决问题？实际上并不行，原因如前面所说是HashMap自身的问题。
那么这个时候我们能做的就是调整代码来解决或者缓解这个问题。
我们之前跟着rust文档学习的最终小demo中是通过线程池来缓解请求量大导致堵塞的问题，那么我们这里能否使用相同的方法呢？实际上确实可以，我们可以开多个HashMap，然后key的存储位置和HashMap的个数关联，这样就能将原来的堵塞问题缓解成1/N。
use bytes::Bytes;
use mini_redis::{Connection, Frame};
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::{Arc, Mutex};
type Db = Arc<Vec<Mutex<HashMap<String, Bytes>>>>;
use tokio::net::{TcpListener, TcpStream};
#[tokio::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
println!("Listening");
let pool_num = 10;
let mut db: Vec<Mutex<HashMap<String, Bytes>>> = Vec::with_capacity(10);
for _ in 0..pool_num {
db.push(Mutex::new(HashMap::new()))
}
let db: Db = Arc::new(db);
loop {
let (socket, _) = listener.accept().await.unwrap();
// Clone the handle to the hash map.
let db = db.clone();
println!("Accepted");
tokio::spawn(async move {
process(socket, db).await;
});
}
}
async fn process(socket: TcpStream, db: Db) {
use mini_redis::Command::{self, Get, Set};
// Connection, provided by `mini-redis`, handles parsing frames from
// the socket
let mut connection = Connection::new(socket);
while let Some(frame) = connection.read_frame().await.unwrap() {
let response = match Command::from_frame(frame).unwrap() {
Set(cmd) => {
let key = cmd.key();
let key_hash = hash(key) as usize;
let db = &db[key_hash % db.len()];
let mut db = db.lock().unwrap();
db.insert(key.to_string(), cmd.value().clone());
Frame::Simple("OK".to_string())
}
Get(cmd) => {
let key = cmd.key();
let key_hash = hash(key) as usize;
let db = &db[key_hash % db.len()];
let db = db.lock().unwrap();
if let Some(value) = db.get(key) {
Frame::Bulk(value.clone())
} else {
Frame::Null
}
}
cmd => panic!("unimplemented {:?}", cmd),
};
// Write the response to the client
connection.write_frame(&response).await.unwrap();
}
}
fn hash (key: &str) -> usize {
let mut hasher = DefaultHasher::new();
key.hash(&mut hasher);
hasher.finish() as usize
}不过这里有个问题，map池的个数初始化就得定下来并且之后绝对不能再改动，因为存取位置和len相关联。
你也可以尝试使用dashmap[19]来解决这个问题
现在我们的server已经差不多了，虽然很简陋。。我们再来实现client。
修改文件夹名字#
在我们开始实现之前，我们先给我们的两个子项目改下名字，一个改为my-redis-server，另一个改为my-redis-client
需要调整的地方有五处，文件夹名字、各自的Cargo.toml，根目录的Cargo.toml。
我们前面了解了channels的基础使用，接下来我们来基于channel实现我们的客户端。
为什么用channel实现客户端#
至于为什么要用channels这个点，我们先来看个例子
use mini_redis::client;
#[tokio::main]
async fn main() {
// Establish a connection to the server
let mut client = client::connect("127.0.0.1:6379").await.unwrap();
// Spawn two tasks, one gets a key, the other sets a key
let t1 = tokio::spawn(async {
let res = client.get("foo").await;
});
let t2 = tokio::spawn(async {
client.set("foo", "bar".into()).await;
});
t1.await.unwrap();
t2.await.unwrap();
}如果不使用通信的话，我们应该第一想法就是上面这种写法，然而上面这种写法是错的，错误的原因自然就是client我们只有一个，而这里有两个异步任务需要拥有它的所有权。那么你可能就会想：那我们给这个client搞一下Arc<Mutex<>>不就好了？
这个想法是没问题的，不过这里的Mutex不能再使用std::sync的了，得改用tokio:sync里的Mutex，因为这里有所有权问题。
use mini_redis::client;
use std::sync::Arc;
use tokio::sync::Mutex;
#[tokio::main]
async fn main() {
// Establish a connection to the server
let client = Arc::new(Mutex::new(client::connect("127.0.0.1:6379").await.unwrap()));
let c1 = client.clone();
let c2 = client.clone();
// Spawn two tasks, one gets a key, the other sets a key
let t1 = tokio::spawn(async move {
let mut guard = (*c1).lock().await;
guard.get("foo").await
});
let t2 = tokio::spawn(async move {
let mut guard = (*c2).lock().await;
guard.set("foo", "bar".into()).await
});
let res1 = t1.await.unwrap();
let res2 = t2.await.unwrap();
dbg!(res1, res2);
} 这个代码并没有问题，跑起来是正常的
但是这么做又导致了一个问题，我们的客户端一次仅处理一个请求也就是说发送给服务器一次仅一个请求，其它都会被堵塞，这就导致了我们server做的优化浪费了。（这么做其实也不是不行，就像是nodeJS和它的事件循环机制，单线程高效,缺点是有上限，当量达到一定的程度，那么就会开始有宏观的堵塞观感了）。
你也可以给每个任务开一个connection，这么做确实是解决了上面这种一次仅处理一个的问题，但是这不理想。
当然，你可以结合这两种效果，设置一个连接数，创建一个连接池。
不过我们还有更优雅的方案：channel。
使用channel，我们可以不用自己去管理锁的逻辑，都交由channel管理，另外也易于拓展，到时候我们可以很快的支持上连接池（多个接收器的场景）。
channel会优雅很多更方便拓展，并且更符合rust异步的开发理念：通过通信去分享数据。
客户端：连接server#
我们先完成对接server的部分，数据暂时先写死。
在开始实现之前，我们有两个问题需要处理：
- 怎么在不同任务间和服务器建立联系
- 拿到数据后怎么送回给之前的任务
第一个问题点自然就是用channel来通信，由receiver也就是接收器来和server通信，这样就不需要去考虑不同任务了。
use bytes::Bytes;
use tokio::sync::mpsc;
#[derive(Debug)]
enum Command {
Get { key: String },
Set { key: String, val: Bytes },
}
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
let manager = tokio::spawn(async move {
use mini_redis::client;
let mut client = client::connect("127.0.0.1:6379").await.unwrap();
while let Some(cmd) = rx.recv().await {
match cmd {
Command::Get { key } => {
let _res = client.get(key.as_str()).await.unwrap();
}
Command::Set { key, val } => {
client.set(key.as_str(), val).await.unwrap();
}
cmd => panic!("the command {:?} is not support!", cmd),
}
}
});
let tx1 = tx.clone();
let tx2 = tx.clone();
let t1 = tokio::spawn(async move {
tx2.send(Command::Set {
key: "test".to_string(),
val: "hi".into(),
})
.await
.unwrap();
});
let t2 = tokio::spawn(async move {
tx1.send(Command::Get {
key: "test".to_string(),
})
.await
.unwrap();
});
t1.await.unwrap();
t2.await.unwrap();
manager.await.unwrap();
}这里我们将receiver也就是rx作为和server连接的接口，所有的provider发送的请求都在rx这里汇聚，然后整理发送给server。
然后我们来实现第二部分，将从server里拿到的数据返回给对应的发射器所在的task。但是这里就有个问题了，由于接收的时候并不是按顺序的，所以这个时候如果按顺序来返回就可能存在问题。
所以我们还需要一种方法支持不同任务间传递数据并且保证数据不会发送错，那就是我们前面提到的oneshot。
如果我们把这个oneshot直接发送给manager，那么返回数据的时候也能原路返回。
那么这个时候原本的数据结构就不够用了，我们需要拓展下字段
type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;
#[derive(Debug)]
enum Command {
Get {
key: String,
resp: Responder<Option<Bytes>>,
},
Set {
key: String,
val: Bytes,
resp: Responder<()>,
},
}多了一个发射器
然后我们来实现这块逻辑
use bytes::Bytes;
use tokio::sync::{mpsc, oneshot};
type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;
#[derive(Debug)]
enum Command {
Get {
key: String,
resp: Responder<Option<Bytes>>,
},
Set {
key: String,
val: Bytes,
resp: Responder<()>,
},
}
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
let manager = tokio::spawn(async move {
use mini_redis::client;
let mut client = client::connect("127.0.0.1:6379").await.unwrap();
while let Some(cmd) = rx.recv().await {
match cmd {
Command::Get { key, resp } => {
let res = client.get(key.as_str()).await;
resp.send(res);
}
Command::Set { key, val, resp } => {
let res = client.set(key.as_str(), val).await;
resp.send(res);
}
cmd => panic!("the command {:?} is not support!", cmd),
}
}
});
let tx1 = tx.clone();
let tx2 = tx.clone();
let t1 = tokio::spawn(async move {
let (rt, rx) = oneshot::channel();
tx2.send(Command::Set {
key: "test".to_string(),
val: "hi".into(),
resp: rt,
})
.await
.unwrap();
let res = rx.await.unwrap();
dbg!(res);
});
let t2 = tokio::spawn(async move {
let (rt, rx) = oneshot::channel();
tx1.send(Command::Get {
key: "test".to_string(),
resp: rt,
})
.await
.unwrap();
let res = rx.await.unwrap();
dbg!(res);
});
t1.await.unwrap();
t2.await.unwrap();
manager.await.unwrap();
}我们在manager拿到server返回的数据之后，通过oneshot::sender将数据发送给oneshot::receiver，这样就拿到对应的返回数据了。
cargo run -p my-redis-server
cargo run -p my-redis-client然后终端正常情况下应该是这样
ok，现在我们打通了和server交流的逻辑。
接下来就是面向客户了。
客户端：提供用户输入接口#
由于教程中并没有实现，所以我这里就简单的用命令行暴露出去了。
use bytes::Bytes;
use std::{io};
use tokio::sync::{mpsc::{self, Sender}, oneshot};
type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;
#[derive(Debug)]
enum Command {
Get {
key: String,
resp: Responder<Option<Bytes>>,
},
Set {
key: String,
val: Bytes,
resp: Responder<()>,
},
}
#[tokio::main]
async fn main() {
loop {
println!("follow rules: [cmd] [key] [value?]");
let mut user_res = String::from("");
io::stdin()
.read_line(&mut user_res)
.expect("unexpect error");
let mut v: Vec<String> = user_res.split_whitespace().map(|s| s.to_string()).collect();
let mut cmd = v[0].clone();
if v.len() < 2 {
println!("your input seem to be missing parameters, check it out");
continue;
}
let key = v[1].clone();
let (tx, mut rx) = mpsc::channel(32);
let manager = tokio::spawn(async move {
use mini_redis::client;
let mut client = client::connect("127.0.0.1:6379").await.unwrap();
while let Some(cmd) = rx.recv().await {
match cmd {
Command::Get { key, resp } => {
let res = client.get(key.as_str()).await;
resp.send(res).unwrap();
}
Command::Set { key, val, resp } => {
let res = client.set(key.as_str(), val).await;
resp.send(res).unwrap();
}
cmd => panic!("the command {:?} is not support!", cmd),
}
}
});
let task = tokio::spawn(async move {
match cmd.as_str() {
"get" => {
spawn_task_get(key, tx).await;
}
"set" => {
let value = v[2].clone();
spawn_task_set(key, value.into(), tx).await;
}
t => {
println!("command {t} is not support!");
}
};
});
task.await.unwrap();
manager.await.unwrap();
}
}
async fn spawn_task_set(key: String, val: Bytes, tx: Sender<Command>) {
let (sd, rc) = oneshot::channel();
let data = Command::Set { key, val, resp: sd };
tx.send(data).await.unwrap();
rc.await;
dbg!("set success");
}
async fn spawn_task_get(key: String, tx: Sender<Command>) {
let (sd, rc) = oneshot::channel();
let data = Command::Get { key, resp: sd };
tx.send(data).await.unwrap();
let res = rc.await;
dbg!(res);
}由于loop会消耗所有权，所以不得已这里将channel创建也放到里面去了。
如果你有更好的方案，可以评论区说一下，谢谢~
client表现
server表现
客户端：实现自己的Connection#
现在我们的connection是基于mini_redis的，我们来运用前面了解到的frame知识自己实现一个。
其实我们现在要做的事情是在造轮子，不过学习嘛~
我们在my-redis-server/src文件夹下创建一个connection.rs文件。
我们的这个Connection首先需要有的自然就是TcpStream，得先拿到buffer再说，然后就是对buffer的操作read和write。另外这里还需要一个类型Frame，这个就是我们前面说到过的。
use bytes::{Bytes, BytesMut};
use tokio::net::TcpStream;
pub type Error = Box<dyn std::error::Error + Send + Sync>;
pub type Result<T> = std::result::Result<T, Error>;
enum Frame {
Simple(String),
Error(String),
Integer(u64),
Bulk(Bytes),
Null,
Array(Vec<Frame>),
}
pub struct Connection {
stream: TcpStream,
buffer: BytesMut,
}
impl Connection {
pub fn new(stream: TcpStream) -> Self {
Connection {
stream,
buffer: BytesMut::with_capacity(4096),
}
}
pub async fn read_frame(&mut self) -> Result<Option<Frame>> {
}
pub async fn write_frame(&mut self, frame: &Frame) -> Result<()> {
}
}我们要做的事情简单地说就是将获取到的buffer转换成frame。
BytesMut：Bytes的mut版本。
接着我们来填充下read_frame的逻辑
pub async fn read_frame(&mut self) -> Result<Option<Frame>> {
loop {
if let Some(frame) = self.parse_frame()? {
return Ok(Some(frame));
}
if 0 == self.stream.try_read_buf(&mut self.buffer).await? {
if self.buffer.is_empty() {
return Ok(None);
} else {
return Err("connection reset by peer".into());
}
}
}
}parse_frame我们等会会实现，这里简单地说就是读buffer，然后写成frame。
这里做的事情很简单，如果parse_frame获取得到帧，那么直接将帧返回出去。如果获取不到并且这里read_buf读到的buffer是0则表明剩余的buffer是空的即读完了，这个时候还需要再确认一波buffer是否为空。
你可能这里会有些疑惑：为什么需要再判断这里的buffer为空？
实际上从stream拿到的是Ok(0)，我们前面多次说过了，这时表明channel关闭了。
而如果这个时候buffer还有字节流，那么就意味着这个frame是partial frame,只是帧的一部分，parse_frame都是一帧一帧读取的。
实际上我们也可以自己来实现这个过程，不过这就需要再拓展一个字段cursor，用于记录当前读的位置。
use tokio::net::TcpStream;
pub struct Connection {
stream: TcpStream,
buffer: Vec<u8>,
cursor: usize,
}
impl Connection {
pub fn new(stream: TcpStream) -> Connection {
Connection {
stream,
// Allocate the buffer with 4kb of capacity.
buffer: vec![0; 4096],
cursor: 0,
}
}
}
pub async fn read_frame(&mut self)
-> Result<Option<Frame>>
{
loop {
if let Some(frame) = self.parse_frame()? {
return Ok(Some(frame));
}
// Ensure the buffer has capacity
if self.buffer.len() == self.cursor {
// Grow the buffer
self.buffer.resize(self.cursor * 2, 0);
}
// Read into the buffer, tracking the number
// of bytes read
let n = self.stream.read(
&mut self.buffer[self.cursor..]).await?;
if 0 == n {
if self.cursor == 0 {
return Ok(None);
} else {
return Err("connection reset by peer".into());
}
} else {
// Update our cursor
self.cursor += n;
}
}
}代码很简单，前面读帧代码一样，后面一段则是在跟踪读到的buffer的位置，cursor光标用于记录当前buffer读到的位置，如果光标所在的位置大于buffer容器的最大值，这个时候就需要扩容，扩大成之前的两倍。
最后判断channel关闭之后光标指向的位置是否是0，如果不是则表示这是一个非正常close。
但我们并不需要这么麻烦，直接使用try_read_buf就好了，它里面会自动跟踪。
然后我们来实现parse_frame部分。
在开始实现之前，我们需要更新下我们的代码，将我们自己实现的Frame替换成mini_redis::frame::Frame，因为我们需要用到里面的几个方法。
+ use mini_redis::frame::Frame;
- // enum Frame {
- // Simple(String),
- // Error(String),
- // Integer(u64),
- // Bulk(Bytes),
- // Null,
- // Array(Vec<Frame>),
- // }parse_frame有两步：
- 确认截取的起始位置
- 截取
而mini_redis::frame::Frame中有提供用于这两步的方法
Frame::checkFrame::parse
然后这还不够，我们还需要借助std::io::Cursor以及bytes::Buf来实现这个过程。
bytes::Buf用于返回当前光标的位置buffer结束位置的字节个数，而Cursor自然就是一个wrapper，它提供了当前读写的position，也就是光标。
use mini_redis::{Frame, Result};
use mini_redis::frame::Error::Incomplete;
use bytes::Buf;
use std::io::Cursor;
fn parse_frame(&mut self)
-> Result<Option<Frame>>
{
// Create the `T: Buf` type.
let mut buf = Cursor::new(&self.buffer[..]);
// Check whether a full frame is available
match Frame::check(&mut buf) {
Ok(_) => {
// Get the byte length of the frame
let len = buf.position() as usize;
// Reset the internal cursor for the
// call to `parse`.
buf.set_position(0);
// Parse the frame
let frame = Frame::parse(&mut buf)?;
// Discard the frame from the buffer
self.buffer.advance(len);
// Return the frame to the caller.
Ok(Some(frame))
}
// Not enough data has been buffered
Err(Incomplete) => Ok(None),
// An error was encountered
Err(e) => Err(e.into()),
}
}如果check判断当前buffer位置离结束位置之间还足够截取至少一个frame，则返回Ok，否则返回Err，而Err存在两种情况，一种是Incomplete表示当前不足以截取一个frame，另一种则是意外断开连接。
而可以截取的场景中我们需要记录下对应的光标位置用于后续判断以及对buffer的截取，然后将光标置为0的位置，因为buffer每次截取过后都会将截取的部分移除，所以下次截取之前光标必须为0，然后调用Frame::parse开始截取转换成frame。最后让buffer自行”了断“被截取的部分。
这里面你可能会有一个问题，这里光标从0开始，怎么知道截取的结束位置？这里的len都没被用到。
我们来稍微看下源码
在分析parse源码之前得先分析下对应的check
起始位置都是有特殊字符的，所以先拿第一个字符用来判断是否是截取的起始位置，然后调用get_line方法 ，这个方法和parse的共用，所以放下面看。
也是以其中一个为栗子，先拿buffer的第一字符用来判断是否是特殊字符，然后get_line开始截取。
len自然就是整个剩余buffer的结束位置，然后开始一个byte一个byte的遍历整个buffer，直到遇到'\r'或者'\n'的特殊结束符才停止，表示截取完一个frame，如果到结束都没有截取完则表明incomplete，也就是从stream读buffer那边还没读完。
读这部分基本是完成了，然后我们来实现写这部分
直接来看下代码
// ...
+ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufWriter};
pub struct Connection {
+ stream: BufWriter<TcpStream>,
buffer: BytesMut,
}
impl Connection {
pub fn new(stream: TcpStream) -> Self {
Connection {
+ stream: BufWriter::new(stream),
buffer: BytesMut::with_capacity(4096),
}
}
// ...
pub async fn write_frame(&mut self, frame: &Frame) -> io::Result<()> {
match frame {
Frame::Array(val) => {
self.stream.write_u8(b'*').await?;
self.write_decimal(val.len() as u64).await?;
for entry in &**val {
self.write_value(entry).await?;
}
}
// The frame type is a literal. Encode the value directly.
_ => self.write_value(frame).await?,
}
// Ensure the encoded frame is written to the socket. The calls above
// are to the buffered stream and writes. Calling `flush` writes the
// remaining contents of the buffer to the socket.
self.stream.flush().await
}
/// Write a frame literal to the stream
async fn write_value(&mut self, frame: &Frame) -> io::Result<()> {
match frame {
Frame::Simple(val) => {
self.stream.write_u8(b'+').await?;
self.stream.write_all(val.as_bytes()).await?;
self.stream.write_all(b"\r\n").await?;
}
Frame::Error(val) => {
self.stream.write_u8(b'-').await?;
self.stream.write_all(val.as_bytes()).await?;
self.stream.write_all(b"\r\n").await?;
}
Frame::Integer(val) => {
self.stream.write_u8(b':').await?;
self.write_decimal(*val).await?;
}
Frame::Null => {
self.stream.write_all(b"$-1\r\n").await?;
}
Frame::Bulk(val) => {
let len = val.len();
self.stream.write_u8(b'$').await?;
self.write_decimal(len as u64).await?;
self.stream.write_all(val).await?;
self.stream.write_all(b"\r\n").await?;
}
// Encoding an `Array` from within a value cannot be done using a
// recursive strategy. In general, async fns do not support
// recursion. Mini-redis has not needed to encode nested arrays yet,
// so for now it is skipped.
Frame::Array(_val) => unreachable!(),
}
Ok(())
}
/// Write a decimal frame to the stream
async fn write_decimal(&mut self, val: u64) -> io::Result<()> {
use std::io::Write;
// Convert the value to a string
let mut buf = [0u8; 12];
let mut buf = Cursor::new(&mut buf[..]);
write!(&mut buf, "{}", val)?;
let pos = buf.position() as usize;
self.stream.write_all(&buf.get_ref()[..pos]).await?;
self.stream.write_all(b"\r\n").await?;
Ok(())
}
}代码挺好理解的，就是将frame按格式转换为buffer然后传输给stream。
这里说下使用到的这个BufWriter，这个东西我们之前异步那块学到过，不过相信大家已经都忘了，简单的看下源码。
注意这里的BufWriter不是原生的，而是tokio实现的异步BufWriter，它实现了AsyncWrite。它在这里的作用是不将byte直接写入到writer里，而是先写到inner的buffer里，等到buffer填充满了再flush给inner的writer，然后清空buffer。
这么做的好处是防止因为数据量太大导致持续性的copy到writer带来较多的损耗，另一方面也可以绕过某些特殊的场景。
你可能会有些疑惑，虽然没有持续性copy到writer里，但是这里持续性的copy到一个buffer不是一样会有较多的损耗？
实际上并不是，只要写入writer，就会触发一系列的改变，而写入buffer就只会触发buffer的改变以及对buffer的监听这两个点。
那么我们这个connection就实现完成了，然后我们接入到my-redis-server/main.rs文件中，重新运行下。
表现正常
底层原理#
rust基础学习--异步day2 - 知乎 (zhihu.com)
我们之前分析过rust异步的原理，这里其实是差不多的，感兴趣的可以回去看下。
贴一张之前分析的图
原理#
在开始之前，我们先来看个非常简单的例子
use tokio::net::TcpStream;
async fn my_async_fn() {
println!("hello from async");
let _socket = TcpStream::connect("127.0.0.1:3000").await.unwrap();
println!("async TCP operation complete");
}
#[tokio::main]
async fn main() {
let what_is_this = my_async_fn();
// Nothing has been printed yet.
what_is_this.await;
// Text has been printed and socket has been
// established and closed.
}这里的my_async_fn()实际上并不是执行这个异步函数，而是返回一个future表示这个异步函数的状态，只有.await触发这个future的时候才会真正的执行函数里的行为。
这里的future指的是std::future::Future这个trait。
我们先来简单的看下Future的源码，注释过多，所以这里直接放代码不截图了。
use std::pin::Pin;
use std::task::{Context, Poll};
pub trait Future {
type Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context)
-> Poll<Self::Output>;
}type Output是关联类型，用于当状态变成Ready的时候返回值。
Pin这个就不用多说了，前面说了挺多次了，有了它，我们可以借用。
和其他语言不同的点是rust中future是运算(computation)本身而不是代表运算发生在后台，rust中future需要自行推进future的状态。
是不是很拗口，简单地说就是你这个什么时候变成ready的逻辑由你自己实现的这个future说了算，当你impl future的时候，你必须实现poll这个方法，而你的是否ready的逻辑就都在这个poll方法里，你还需要返回Pending或者Ready的状态才行。
而你这个future自身还需要去通知executor来执行自己才行(wake)，这就是为什么rust里future的运算是它自身的原因。
我们来简单的实现一个
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
struct Delay {
when: Instant,
}
impl Future for Delay {
type Output = &'static str;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
-> Poll<&'static str>
{
if Instant::now() >= self.when {
println!("Hello world");
Poll::Ready("done")
} else {
// Ignore this line for now.
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
#[tokio::main]
async fn main() {
let when = Instant::now() + Duration::from_millis(10);
let future = Delay { when };
let out = future.await;
assert_eq!(out, "done");
}代码很简单，就是延迟一段时间输出Hello world。
这里你唯一可能不理解的地方就是cx.waker().wake_by_ref()。cx自然就是Context也就是上下文环境，它里面有很多东西，我们这里只需要这个waker。waker()代码不用看了，就是拿到对应的waker，而wake_by_ref则是通知执行器(executor)去触发future的poll。
那么这个waker是什么呢？这里简单的说就是future用来主动通知executor去触发自己的poll的工具，如果不通知executor，那么你这个future就会一直停留在Pending状态。
我们来试下注释掉cx.waker().wake_by_ref()这行代码，得到的结果如下
我在注释的地方加了一行打印pending的代码，可以看到这里只触发了一次future.poll。
如果我们注释回cx.waker()这一行，然后不移除打印pending的代码，这个时候会pending好几次直到达到延迟时间。
不过从原理的角度来看，这个wake()触发之后和executors之间又发生了什么事我们并不清楚，甚至是executors我们都没有接触到。我们只知道是wake()通知了executors去触发future.poll。
我们来简单的实现一个微型tokio来理解以及加深这块的印象
mini tokio#
首先我们引入一下futures的crate
[dependencies]
tokio = { version = "1", features = ["full"] }
mini-redis = "0.4"
tokio-stream = "0.1"
futures = "0.3"然后直接贴下代码
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use std::collections::VecDeque;
use futures::task;
struct Delay {
when: Instant,
}
impl Future for Delay {
type Output = &'static str;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
-> Poll<&'static str>
{
if Instant::now() >= self.when {
println!("Hello world");
Poll::Ready("done")
} else {
// Ignore this line for now.
print!("pending");
// cx.waker().wake_by_ref();
Poll::Pending
}
}
}
fn main() {
let mut mini_tokio = MiniTokio::new();
mini_tokio.spawn(async {
let when = Instant::now() + Duration::from_millis(10);
let future = Delay { when };
let out = future.await;
assert_eq!(out, "done");
});
mini_tokio.run();
}
struct MiniTokio {
tasks: VecDeque<Task>,
}
type Task = Pin<Box<dyn Future<Output = ()> + Send>>;
impl MiniTokio {
fn new() -> MiniTokio {
MiniTokio {
tasks: VecDeque::new(),
}
}
/// Spawn a future onto the mini-tokio instance.
fn spawn<F>(&mut self, future: F)
where
F: Future<Output = ()> + Send + 'static,
{
self.tasks.push_back(Box::pin(future));
}
fn run(&mut self) {
let waker = task::noop_waker();
let mut cx = Context::from_waker(&waker);
while let Some(mut task) = self.tasks.pop_front() {
if task.as_mut().poll(&mut cx).is_pending() {
self.tasks.push_back(task);
}
}
}
}这块代码很简单，就是用队列将任务存储起来，然后轮询去判断它们返回的状态。
这里我们的miniTokio就是一个executors.
不过这里还没有完成，你会发现这里的waker一点用没有，我们把之前的cx.waker().wake_by_ref()给注释掉也是一样一堆pending最后来一个hello world。
接着我们来实现waker这一部分。
不过在开始之前，我们有个点需要注意：我们现在每次都在wake，这是不对的，这么做和没有waker的存在是一样的效果，都会导致executors执行N次。
实际上延迟器并不需要每次都去wake，实际只需要触发两次，一次是初始阶段，第二次是等待N秒之后通过wake去通知executors再触发自己一次。
基于这一点，我们来修改下我们的代码
// ...
+ use std::thread;
struct Delay {
when: Instant,
}
impl Future for Delay {
type Output = &'static str;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
-> Poll<&'static str>
{
if Instant::now() >= self.when {
println!("Hello world");
Poll::Ready("done")
} else {
// Get a handle to the waker for the current task
let waker = cx.waker().clone();
let when = self.when;
// Spawn a timer thread.
thread::spawn(move || {
let now = Instant::now();
if now < when {
thread::sleep(when - now);
}
waker.wake();
});
Poll::Pending
}
}
}
// ...我们直接开了一个线程，让里面等待N秒之后再触发wake。
ok，我们现在来实现连接executors那部分。
那么怎么让waker主动通知到executors去触发future.poll呢？根据我们以往的知识，这里用channel再合适不过。
发送什么给executors呢？当然是future，将整个future发送给executors，甚至都不需要id去匹配等。
不过这里就有几个问题：
- 线程
Send的问题，我们需要传递的东西必须要Send + Sync才行，而我们标准库里的的channel差一个Sync。 executors保留一个receiver,而其它future共用一个Sender，我们可以用Arc包裹。不过这里还有个问题，怎么送，怎么把Sender和future关联。回想下我们是怎么创建task的，我们通过tokio::spawn的方式创建task，那么在这里我们可以搞一个task用来包裹future和Sender。
这里就需要我们再引入一个crate：
crossbeam = "0.8"然后我们先来调整下executors的部分代码
use crossbeam::channel;
use std::sync::Arc;
struct MiniTokio {
scheduled: channel::Receiver<Arc<Task>>,
sender: channel::Sender<Arc<Task>>,
}scheduled：调度器，实际上就是接收channel传过来的task。sender：发射器，waker最关键的部分。不过现在只能是放我们这里，因为所有权等问题。
然后我们来实现task
use std::sync::{Arc, Mutex};
struct Task {
// The `Mutex` is to make `Task` implement `Sync`. Only
// one thread accesses `future` at any given time. The
// `Mutex` is not required for correctness. Real Tokio
// does not use a mutex here, but real Tokio has
// more lines of code than can fit in a single tutorial
// page.
future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
executor: channel::Sender<Arc<Task>>,
}
impl Task {
fn schedule(self: &Arc<Self>) {
self.executor.send(self.clone());
}
}task自身带有future这个不用说。另一个则是Sender，这样future和Sender就建立联系了，进而executors和future也建立联系了。
schedule方法不用多说。
还有一点，这里用Mutex包裹这个future实属无奈，用于获取Sync，由于future自身没有Sync，那么跨线程就无法传递数据。
然后我们来给Task实现Wake，这里的Wake直接采用futures::task::ArcWake。
use futures::task::{self, ArcWake};
use std::sync::Arc;
impl ArcWake for Task {
fn wake_by_ref(arc_self: &Arc<Self>) {
arc_self.schedule();
}
}这里自然不必多说。你可能会觉得这里的waker没必要，我们直接调用schedule方法即可，这里看确实没这个必要，我们后面就会知道为什么有必要了。
然后我们来打通task和executors的联系。
impl MiniTokio {
fn run(&self) {
while let Ok(task) = self.scheduled.recv() {
task.poll();
}
}
/// Initialize a new mini-tokio instance.
fn new() -> MiniTokio {
let (sender, scheduled) = channel::unbounded();
MiniTokio { scheduled, sender }
}
/// Spawn a future onto the mini-tokio instance.
///
/// The given future is wrapped with the `Task` harness and pushed into the
/// `scheduled` queue. The future will be executed when `run` is called.
fn spawn<F>(&self, future: F)
where
F: Future<Output = ()> + Send + 'static,
{
Task::spawn(future, &self.sender);
}
}
impl Task {
fn poll(self: Arc<Self>) {
// Create a waker from the `Task` instance. This
// uses the `ArcWake` impl from above.
let waker = task::waker(self.clone());
let mut cx = Context::from_waker(&waker);
// No other thread ever tries to lock the future
let mut future = self.future.try_lock().unwrap();
// Poll the future
let _ = future.as_mut().poll(&mut cx);
}
// Spawns a new task with the given future.
//
// Initializes a new Task harness containing the given future and pushes it
// onto `sender`. The receiver half of the channel will get the task and
// execute it.
fn spawn<F>(future: F, sender: &channel::Sender<Arc<Task>>)
where
F: Future<Output = ()> + Send + 'static,
{
let task = Arc::new(Task {
future: Mutex::new(Box::pin(future)),
executor: sender.clone(),
});
let _ = sender.send(task);
}
}MiniTokio::run自然就是等待接收channel发送过来的数据，然后处理。MiniTokio::new这个就不必多说,这也是为什么我们需要先把channel的Sender放到MiniTokio里面的原因之一。MiniTokio::spawn这个方法调用Task::spawn创建task的同时将future和sender保存到task里，这样future和sender就建立联系了。Task::poll注意这个方法不是Future::poll，Future::poll是用于转变future的状态，而Task::poll则是获取future，然后执行future::poll同时将上下文环境的waker传递给这个future.poll。Task::spawn：这个不必多说，就是在创建task，将future和Sender建立联系。
那么现在整体都打通了，我们来跑一下试下
可以看到pending只触发了一次
最后我们来优化下我们的代码，我们前面delay这块的代码实现future的时候poll里面每次都会去创建一个线程，这里我们虽然只跑了一次，但是如果这个future被转移到别的task里面呢？rust是支持这么做的，那么就可能存在转移之后又去触发一次poll，那么这个时候可能还没到延迟的时间，结果就是又触发一次线程创建。
我们来看个例子：
use futures::future::poll_fn;
use std::future::Future;
use std::pin::Pin;
#[tokio::main]
async fn main() {
let when = Instant::now() + Duration::from_millis(10);
let mut delay = Some(Delay { when });
poll_fn(move |cx| {
let mut delay = delay.take().unwrap();
let res = Pin::new(&mut delay).poll(cx);
assert!(res.is_pending());
tokio::spawn(async move {
delay.await;
});
Poll::Ready(())
}).await;
}我们先是通过poll_fn方法创建了一个task，然后给这个delay的future给poll了，接着将它转移到了tokio::spawn出来的task里面，然后里面又去触发.await，这样就触发了两次。
那么我们要怎么防止这种情况发生呢？
这个时候就轮到我们前面说的没必要的waker出场了，我们可以锁waker，将waker和Delay关联起来
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use futures::task::{self, ArcWake};
use futures::future::poll_fn;
use std::thread;
use std::sync::{Arc, Mutex};
use crossbeam::channel;
use std::task::{Waker};
struct Delay {
when: Instant,
// This is Some when we have spawned a thread, and None otherwise.
waker: Option<Arc<Mutex<Waker>>>,
}
impl Future for Delay {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
if let Some(waker) = &self.waker {
let mut waker = waker.lock().unwrap();
if !waker.will_wake(cx.waker()) {
*waker = cx.waker().clone();
}
} else {
let when = self.when;
let waker = Arc::new(Mutex::new(cx.waker().clone()));
self.waker = Some(waker.clone());
thread::spawn(move || {
let now = Instant::now();
if now < when {
thread::sleep(when - now);
}
let waker = waker.lock().unwrap();
waker.wake_by_ref();
});
}
if Instant::now() >= self.when {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
#[tokio::main]
async fn main() {
let when = Instant::now() + Duration::from_millis(10);
let mut delay = Some(Delay { when, waker: None });
poll_fn(move |cx| {
let mut delay = delay.take().unwrap();
let res = Pin::new(&mut delay).poll(cx);
assert!(res.is_pending());
tokio::spawn(async move {
delay.await;
});
Poll::Ready(())
}).await;
}这里做的事情很简单，就是当已经有waker了的时候，就不再去触发线程创建了。
不过注意要同步最新的waker，毕竟一个是前朝官员，哪天人就不见了。
不过waker这玩意儿是较底层的东西，我们一般不会把它和我们的业务代码关联，那么这个时候怎么办呢？
实际上这里还有一个tokio::sync::Notify的东西，它可以自动帮我们完成上面的行为。
use tokio::sync::Notify;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::thread;
async fn delay(dur: Duration) {
let when = Instant::now() + dur;
let notify = Arc::new(Notify::new());
let notify2 = notify.clone();
thread::spawn(move || {
let now = Instant::now();
if now < when {
thread::sleep(when - now);
}
notify2.notify_one();
});
notify.notified().await;
}我们通过notify_one通知还在pending中的task的waker可以去wake了。
补充#
并发(concurrency)和并行(parallelism)的区别#
虽然这俩我们平时都当作一样，因为都是需要处理多个任务，所以自然而然就合并到一起去了。
但其实这俩还是有区别的，举个栗子：
- 并发：一人交换着完成两个任务，一个任务遇到了需要等待的功能就先挂起，然后去处理另一个任务。
- 并行：两个任务各自被分配给不同的两个人。
而tokio则是并发的，并不需要去使用传统的方式开多个线程（有点类似于nodeJs的事件循环机制？）。
async block is not Send#
前面说过tokio底层是基于多线程调度器的，这就意味着你可能每次.await都是在不同的线程里，那么这个时候你的数据就一定需要确保实现了Send这个trait，比如下面这段代码
async fn spawn () {
tokio::spawn(async move {
let db: Arc<Mutex<HashMap<String, String>>> = Arc::new(Mutex::new(HashMap::new()));
let m = db.clone();
increment_and_do_stuff(&m).await;
});
}
async fn increment_and_do_stuff(mutex: &Mutex<HashMap<String, String>>) {
let mut lock: MutexGuard<HashMap<String, String>> = mutex.lock().unwrap();
(*lock).insert("a".to_string(), "b".to_string());
do_something_async().await;
} // lock goes out of scope here
async fn do_something_async () {
} 乍一看是对的，实际上则是会报错
报错原因自然就是前面说到的，这个spawn出来的task在遇到.await可能会被转移到别的线程里面去，那么这个时候这个数据没有实现Send这个trait就会报错。
这种情况也好解决，只要这个字段在遇到的.await之前被drop即可。
async fn increment_and_do_stuff(mutex: &Mutex<HashMap<String, String>>) {
{
let mut lock: MutexGuard<HashMap<String, String>> = mutex.lock().unwrap();
(*lock).insert("a".to_string(), "b".to_string());
}
do_something_async().await;
}
这样就正常了。
或者更优雅一点
use std::sync::Mutex;
struct CanIncrement {
mutex: Mutex<i32>,
}
impl CanIncrement {
// This function is not marked async.
fn increment(&self) {
let mut lock = self.mutex.lock().unwrap();
*lock += 1;
}
}
async fn increment_and_do_stuff(can_incr: &CanIncrement) {
can_incr.increment();
do_something_async().await;
}最后你还可以直接使用tokio提供的mutex
use tokio::sync::Mutex; // note! This uses the Tokio mutex
// This compiles!
// (but restructuring the code would be better in this case)
async fn increment_and_do_stuff(mutex: &Mutex<i32>) {
let mut lock = mutex.lock().await;
*lock += 1;
do_something_async().await;
} // lock goes out of scope here 这样完全不会报错，不过代价自然就是更多的损耗。
总结#
好段时间没碰rust了，什么都忘了。。。。。。
参考#
- ^tokio https://tokio.rs/
- ^futures https://crates.io/crates/futures
- ^rayon https://docs.rs/rayon/
- ^spawning https://tokio.rs/tokio/tutorial/spawning
- ^channels https://tokio.rs/tokio/tutorial/channels
- ^mpsc https://docs.rs/tokio/1/tokio/sync/mpsc/index.html
- ^oneshot https://docs.rs/tokio/1/tokio/sync/oneshot/index.html
- ^broadcast https://docs.rs/tokio/1/tokio/sync/broadcast/index.html
- ^watch https://docs.rs/tokio/1/tokio/sync/watch/index.html
- ^async-channel https://docs.rs/async-channel/
- ^std::sync::mpsc https://doc.rust-lang.org/stable/std/sync/mpsc/index.html
- ^crossbeam::channel https://docs.rs/crossbeam/latest/crossbeam/channel/index.html
- ^I/O https://tokio.rs/tokio/tutorial/io
- ^select https://tokio.rs/tokio/tutorial/select
- ^promise.race https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/race
- ^streams https://tokio.rs/tokio/tutorial/streams
- ^setup https://tokio.rs/tokio/tutorial/setup
- ^redis https://redis.io/
- ^dashmap https://docs.rs/dashmap
发布于 2023-04-27 15:06・IP 属地广东