读写数据

读写数据

非阻塞I/O

在概述我们简单提到的是tokio的I/O类型实现了无阻塞异步版std:![:io:](https://www.emoji-cheat-sheet.com/graphics/emojis/io.png ":io:"):Read和std:![:io:](https://www.emoji-cheat-sheet.com/graphics/emojis/io.png ":io:"):Write，名为 AsyncRead和AsyncWrite。这些是Tokio的I/O中不可或缺的一部分，在使用I/O代码时，理解这些东东非常重要。

注意：在本节中，我们将主要讨论AsyncRead， AsyncWrite几乎完全相同，只是将数据写入I/O资源（如TCP套接字）而不是从中读取。

好，让我们来看看，AsyncRead能干什么：

1. use std::io::Read;
2. pub trait AsyncRead: Read {

6. }

嗯，这都说了些啥？嗯，AsyncRead只是继承了std::io中的Read，以及一份额外的契约。AsyncRead文档中提到：

此trait继承自std::io::Read并表明I/O对象是非阻塞的。当无可用数据时，所有非阻塞I/O对象都必须返回一个error而不是阻塞当前线程。

最后一部分至关重要。如果你为一个类型实现AsyncRead，你得保证调用read而不会阻塞。相反，如果它不是非阻塞的，则应该返回io:![:ErrorKind:](https://www.emoji-cheat-sheet.com/graphics/emojis/ErrorKind.png ":ErrorKind:"):WouldBlock错误以表明操作将被阻塞（例如因为没有可用的数据）。poll_read方法依赖于此：

1. fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, std::io::Error> {
2. match self.read(buf) {
3. Ok(t) => Ok(Async::Ready(t)),
4. Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
5. Ok(Async::NotReady)
6. }
7. Err(e) => Err(e),
8. }
9. }

这段代码应该很熟悉。如果仔细看一眼，poll_read 看起来很像Future::poll。那是因为它基本上就是！一个实现AsyncRead的类型本质上就是一个可以尝试从中读取数据的Future，它将通知你是否Ready（某些数据已被读取）或NotReady（你需要再次poll_read）。

使用I/O Future

由于AsyncRead（和AsyncWrite）几乎都是Futures，你可以很容易地将它们嵌入到你自己的Futures中，poll_read它们，就像poll任何其他嵌入Future一样。您甚至可以根据需要使用try_ready!，这个micro可以传递error和NotReady状态。我们将在下一节中更多地讨论如何直接使用这些traits。但是，在许多情况下，为了简化，Tokio在tokio::io中提供了许多有用的组合器combinator，用于在AsyncRead和AsyncWrite之上执行常见的I/O操作。通常，它们封装了AsyncRead或AsyncWrite类型，实现了Future，并在给定的读或写操作完成时完成。

第一个好用的I/O组合器是read_exact。它需要一个可变的buffer（&mut [u8]）和AsyncRead的实现作为参数，并返回一个Future读取足够的字节来填充buffer。在内部，返回的future只是跟踪它已经读取了多少字节，并在AsyncRead时继续发出poll_ready （如果需要，返回NotReady），直到它正好填满了buffer。那时，它返回带着填满的buffer的Ready(buf)。让我们来看看：

1. use tokio::net::tcp::TcpStream;
2. use tokio::prelude::*;

4. let addr = "127.0.0.1:12345".parse().unwrap();
5. let read_8_fut = TcpStream::connect(&addr)
6. .and_then(|stream| {

11. let mut buf = vec![0; 8];

15. tokio::io::read_exact(stream, buf)
16. })
17. .inspect(|(_stream, buf)| {

21. println!("got eight bytes: {:x?}", buf);
22. });

第二个常用的I/O组合器是write_all。它需要一个buffer（&[u8]）和一个AsyncWrite的实现作为参数，并返回一个将缓冲区的所有字节用poll_write写入AsyncWrite的future。当Future被resolve，整个缓冲区已经写完并被刷新。我们可以结合 read_exact使用，来echo服务器的任何内容：

1. use tokio::net::tcp::TcpStream;
2. use tokio::prelude::*;

4. let echo_fut = TcpStream::connect(&addr)
5. .and_then(|stream| {

8. let mut buf = vec![0; 32];

10. tokio::io::read_exact(stream, buf)
11. })
12. .and_then(|(stream, buf)| {

14. tokio::io::write_all(stream, buf)
15. })
16. .inspect(|(_stream, buf)| {
17. println!("echoed back {} bytes: {:x?}", buf.len(), buf);
18. });

Tokio还带有一个I/O组合器来实现上面例子中这种复制。它（或许不足为奇）被称为copy。copy接受一个AsyncRead和一个AsyncWrite，连续地将从中AsyncRead读出的所有字节，写入到AsyncWrite，直到 poll_read指示输入已经关闭并且所有字节都已写出并刷新到输出。这是我们在echo服务器中使用的组合器！它大大简化了我们上面的示例，并使其适用于任何量级的服务器数据！

1. use tokio::net::tcp::TcpStream;
2. use tokio::prelude::*;

4. let echo_fut = TcpStream::connect(&addr)
5. .and_then(|stream| {

9. let (reader, writer) = stream.split();

12. tokio::io::copy(reader, writer)
13. })
14. .inspect(|(bytes_copied, r, w)| {
15. println!("echoed back {} bytes", bytes_copied);
16. });

简约！

到目前为止我们谈到的组合器都是用于相当底层的操作：读取字节，写入字节，复制字节。但是，通常情况下，您希望在更高级别的表述上操作，例如“lines”。这些Tokio也帮你搞定了！lines接受一个 AsyncRead，并返回一个Stream，从输入中产生yield每一行，直到没有更多行要读：

1. use tokio::net::tcp::TcpStream;
2. use tokio::prelude::*;

4. let lines_fut = TcpStream::connect(&addr).and_then(|stream| {

12. let stream = std::io::BufReader::new(stream);
13. tokio::io::lines(stream).for_each(|line| {
14. println!("server sent us the line: {}", line);

20. Ok(())
21. })
22. });

在tokio::io中，还有更多的I/O组合器，在你决定自己写一个之前，不妨先看一下是否已经有了实现！

拆分I/O资源

上面的copy和echo服务器例子中都包含下面这个神秘的代码片段：

1. let (reader, writer) = socket.split();
2. let bytes_copied = tokio::io::copy(reader, writer);

正如上面的注释所解释的那样，我们将TcpStream（socket）拆分为读半部分和写半部分，并使用我们上面讨论的copy组合器产生一个Future异步复制从读半部分到写半部分的所有数据。但究竟为什么需要这种“split”呢？毕竟，AsyncRead::poll_read和 AsyncWrite::poll_write只接受参数&mut self。

要回答这个问题，我们需要回顾一下Rust的Ownership机制。回想一下，Rust只允许你在任一时刻对一给定变量拥有单个可变引用。但是我们必须传递两个参数给copy，一个用于从哪里读，另一个指定写到哪里。但是，一旦我们将一个可变引用作为其中一个参数传递给TcpStream，我们就不能构造第二个指向它的可变引用作为第二个参数传递给它！我们知道，copy 将不能同时读取和写入到其参数，但是这并没有在copy的类型定义中表示出来。

进入split方法，一个AsyncRead中提供的方法也实现了AsyncWrite。如果我们看一下方法签名，我们就会看到：

1. fn split(self) -> (ReadHalf<Self>, WriteHalf<Self>)
2. where Self: AsyncWrite { ... }

返回的ReadHalf实现了AsyncRead，WriteHalf 实现了AsyncWrite。至关重要的是，我们现在有两个独立的 指针在我们的类型中，我们可以单独传递它们。这很方便copy，但这也意味着我们可以将每一半传递到不同的Future，并完全独立地处理读写操作！在幕后，split确保如果我们同时尝试读写，一次只发生其中一个。

传输

在I/O应用中，将一个AsyncRead转换为Stream（像lines这样）或将一个AsyncWrite转化为Sink相当普遍。他们经常想要把从网络上读、写字节的方式进行抽象，并让大多数应用程序代码处理更方便的“request”和“response”类型。这通常被称为“framing”：您可以将它们视为接收和发送的应用程序数据的“帧”，而不仅仅视您的连接视为字节进、字节出。帧化的字节流通常被称为“传输”。

传输通常使用编解码器codec实现。例如， lines表述了一个非常简单的编解码器，用换行符\n分割字节字符串，并在将其传递给应用程序之前，将每个帧作为字符串解析。Tokio在tokio::codec中提供了helpers来帮助实现新的编解码器; 你为你的传输实现了Encoder和 Decoder traits，并用Framed::new从你的字节流中创建一个Sink + Stream（比如一个TcpStream）。这几乎就像魔术一样！有些编解码器只用读或写端（比如lines）。让我们来看一下编写基于行的编解码器的简单实现（即使LinesCodec 已经存在）：

1. extern crate bytes;
2. use bytes::{BufMut, BytesMut};
3. use tokio::codec::{Decoder, Encoder};
4. use tokio::prelude::*;

8. struct LinesCodec;

11. fn bad_utf8<E>(_: E) -> std::io::Error {
12. std::io::Error::new(std::io::ErrorKind::InvalidData, "Unable to decode input as UTF8")
13. }

18. impl Encoder for LinesCodec {
19. type Item = String;
20. type Error = std::io::Error;

22. fn encode(&mut self, line: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> {

27. buf.reserve(line.len() + 1);

29. buf.put(line);
30. buf.put_u8(b'\n');
31. Ok(())
32. }
33. }

41. impl Decoder for LinesCodec {
42. type Item = String;
43. type Error = std::io::Error;

46. fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
47. Ok(if let Some(offset) = buf.iter().position(|b| *b == b'\n') {

50. let mut line = buf.split_to(offset + 1);

52. Some(
53. std::str::from_utf8(&line[..line.len() - 1])
54. .map_err(bad_utf8)?
55. .to_string(),
56. )
57. } else {

60. None
61. })
62. }

65. fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
66. Ok(match self.decode(buf)? {
67. Some(frame) => {

69. Some(frame)
70. },
71. None => {

75. if buf.is_empty() {
76. None
77. } else {
78. Some(
79. std::str::from_utf8(&buf.take()[..])
80. .map_err(bad_utf8)?
81. .to_string(),
82. )
83. }
84. }
85. })
86. }
87. }