[TOC] ### `Future`特點（trait） `Future`特點的定義如下： ``` trait Future { /// The type of the value returned when the future completes. type Item; /// The type representing errors that occurred while processing the computation. type Error; /// The function that will be repeatedly called to see if the future is /// has completed or not fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error>; } ``` #### Item和Error `Item`是`Future`在完成時將產生的值的類型，錯誤是如果在導致`Future`能夠完成之前出現錯誤，`Future`可能會產生的錯誤類型。 #### Poll方法 `poll`是在`tokio`運行時調用的，以便查看Future是否已完成。 如果你很好奇：Async是一個帶有值的枚舉，Ready(Item)或者NotReady告訴`tokio`運行時`Future`是否完成。 ***** ### 流(Streams) 流是同類`Future`的迭代器。流不會在未來的某個時間點產生值，而是在將來的某個時刻產生一組值。換句話說，像`Future`一樣，流在未來的某一點上不會產生一個值。他們隨著時間的推移繼續產生值。 Streams在實現過程中非常類似于future： ``` trait Stream { /// The type of the value yielded by the stream. type Item; /// The type representing errors that occurred while processing the computation. type Error; /// The function that will be repeatedly called to see if the stream has /// another value it can yield fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>; } ``` ### Futures是用來管理異步邏輯的基本構件 #### EchoServer的實現 #### 創建項目echo_server 生成一個新的項目 ``` $ cargo new echo_server $ cd echo_server ``` 在`Cargo.toml`中添加依賴項: ``` [dependencies] tokio = "0.1" futures = "0.1" ``` 文件src/main.rs: ``` extern crate tokio; extern crate futures; use tokio::io; use tokio::net::TcpListener; use tokio::prelude::*; ``` #### 服務器必要結構： - 綁定`TcpListener`到本地端口。 - 定義接受入站連接并處理它們的任務。 - 生成服務器任務。 - 啟動TokioRuntime #### 編寫一個什么都不做的服務器 ``` let addr = "127.0.0.1:7878".parse().unwrap(); let listener = TcpListener::bind(&addr).unwrap(); let _server = listener.incoming().for_each(|socket| { Ok(()) }).map_err(|err|{ println!("accept error:{:?}", err) }); tokio::run(_server); ``` `TcpListener`是監聽器,`incoming`將監聽器轉換為入站客戶端連接流，用`for_each`將產生每個入站客戶端連接。 目前我們沒有對此入站連接做任何事情. 一旦我們擁有了我們的服務器，我們就可以將它交給`tokio::run`。到目前為止，我們的服務器功能一無所獲。由Tokio運行時驅動我們的`Future`完成。 ` ` 注意：我們必須在服務器上調用`map_err`，因為`tokio :: run`需要一個`Item`為type（）和`Error`為type（）的`Future`。 這是為了確保在將`Future`交付給運行時之前處理所有值和錯誤。 ` ` #### 處理連接 將從套接字讀取的所有數據復制回套接字本身，使用`io::copy`函數，他需要兩個參數，但是我們只有一個參數socket，但是socket有一個函數`split`,它將可讀和可寫的流分成兩半。 然后，`copy`函數返回一個`Future`，當復制操作完成時，將接收此`Future`，解析為復制的數據量: ``` let server = listener.incoming().for_each(|socket| { let (reader, writer) = socket.split(); let amount = io::copy(reader, writer); let msg = amount.then(|result| { match result { Ok((amount, _, _)) => println!("wrote {} bytes", amount), Err(e) => println!("error: {}", e), } Ok(()) }); tokio::spawn(msg); Ok(()) }) ``` #### tokio::spawn [tokio::spawn](https://docs.rs/tokio-executor/0.1/tokio_executor/fn.spawn.html)的調用是關鍵所在,至關重要的是我們希望所有`clients`同時取得進展，而不是在完成另一個`client`時阻止其中一個。 為此，使用`tokio :: spawn`函數在后臺執行工作。 如果我們沒有這樣做，那么`for_each`中塊的每次調用都會在一次解決，這意味著我們永遠不會同時處理兩個客戶端連接！ ***** ###