任务
- 消息传递
- 任务通知

任务

任务是应用程序的“逻辑单元”。它们类似于Go的goroutine和Erlang的进程，但是异步。换句话说，任务是异步绿色线程。

鉴于任务运行异步逻辑位，它们由Future特征表示。任务完成处理后，任务的future实现将以（）值完成。

任务被传递给执行程序，执行程序处理任务的调度。执行程序通常在一组或一组线程中调度许多任务。 任务不得执行计算繁重的逻辑，否则将阻止其他任务执行 。因此，不要尝试将斐波那契序列计算为任务。

任务通过直接实施Future特征或通过使用future和tokio crate中可用的各种组合函数构建future来实现。

下面是一个使用HTTP get从URI获取值并缓存结果的示例。

检查缓存以查看是否存在URI条目。
如果没有条目，请执行HTTP get。
将响应存储在缓存中。
返回response。整个事件序列也包含超时，以防止无限制的执行时间。

8. fn cache_get(uri: &str)
9. -> Box<Future<Item = Option<String>, Error = Error>>
10. { ... }

12. fn cache_put(uri: &str, val: String)
13. -> Box<Future<Item = (), Error = Error>>
14. { ... }

17. fn http_get(uri: &str)
18. -> Box<Future<Item = String, Error = Error>>
19. { ... }

21. fn fetch_and_cache(url: &str)
22. -> Box<Future<Item = String, Error = Error>>
23. {

27. let url = url.to_string();

29. let response = http_get(&url)
30. .and_then(move |response| {
31. cache_put(&url, response.clone())
32. .map(|_| response)
33. });

35. Box::new(response)
36. }

38. let url = "https://example.com";

40. let response = cache_get(url)
41. .and_then(|resp| {

45. match resp {
46. Some(resp) => Either::A(future::ok(resp)),
47. None => {
48. Either::B(fetch_and_cache(url))
49. }
50. }
51. });

57. let task = Timeout::new(response, Duration::from_secs(20));

59. my_executor.spawn(task);

由于这些步骤对于完成任务都是必需的，因此将它们全部分组到同一任务中是有意义的。

但是，如果不是在缓存未命中时更新缓存，而是希望在一个时间间隔内更新缓存值，那么将其拆分为多个任务是有意义的，因为这些步骤不再直接相关。

2. let url = "https://example.com";

5. let update_cache = Interval::new(Duration::from_secs(60))

8. .for_each(|_| {
9. fetch_and_cache(url)
10. .map(|resp| println!("updated cache with {}", resp))
11. });

14. my_executor.spawn(update_cache);

18. let response = cache_get(url);
19. let task = Timeout::new(response, Duration::from_secs(20));

21. my_executor.spawn(task);

消息传递

就像Go和Erlang一样，任务可以使用消息传递进行通信。实际上，使用消息传递来协调多个任务是很常见的。这允许独立任务仍然相互作用。

future包提供了一个同步模块，其中包含一些适合跨任务传递消息的通道类型。

oneshot是一个用于发送一个值的通道。
mpsc是用于发送许多（零个或多个）值的通道。前面的例子并不完全正确。鉴于任务同时执行，无法保证缓存更新任务在其他任务尝试从缓存中读取时将第一个值写入缓存。

这是使用消息传递的完美情况。高速缓存更新任务可以发送消息，通知其他任务它已使用初始值启动了高速缓存。

1. let url = "https://example.com";

3. let (primed_tx, primed_rx) = oneshot::channel();

5. let update_cache = fetch_and_cache(url)

7. .then(|_| primed_tx.send(()))

9. .then(|_| {
10. Interval::new(Duration::from_secs(60))
11. .for_each(|_| {
12. fetch_and_cache(url)
13. .map(|resp| println!("updated cache with {}", resp))
14. })
15. });

18. my_executor.spawn(update_cache);

21. let response = primed_rx
22. .then(|_| cache_get(url));

24. let task = Timeout::new(response, Duration::from_secs(20));

26. my_executor.spawn(task);

任务通知

使用Tokio构建的应用程序被构造为一组并发运行的任务。这是服务器的基本结构：

1. let server = listener.incoming().for_each(|socket| {

3. tokio::spawn(process(socket));

5. Ok(())
6. })
7. .map_err(|_| ());

9. tokio::run(server);

在这种情况下，我们为每个入站服务器套接字生成一个任务。但是，也可以实现处理同一套接字上所有入站连接的服务器future：

1. pub struct Server {
2. listener: TcpListener,
3. connections: Vec<Box<Future<Item = (), Error = io::Error> + Send>>,
4. }

6. impl Future for Server {
7. type Item = ();
8. type Error = io::Error;

10. fn poll(&mut self) -> Result<Async<()>, io::Error> {

12. loop {
13. match self.listener.poll_accept()? {
14. Async::Ready((socket, _)) => {
15. let connection = process(socket);
16. self.connections.push(connection);
17. }
18. Async::NotReady => break,
19. }
20. }

23. let len = self.connections.len();

25. for i in (0..len).rev() {
26. match self.connections[i].poll()? {
27. Async::Ready(_) => {
28. self.connections.remove(i);
29. }
30. Async::NotReady => {}
31. }
32. }

36. Ok(Async::NotReady)
37. }
38. }

这两种策略在功能上是等效的，但具有明显不同的运行时特性。

通知发生在任务级别。该任务不知道哪个子future触发了通知。因此，无论何时轮询任务，都必须尝试轮询所有子future。

在此任务中，有三个子future可以进行轮询。如果其中一个子future所包含的资源转为“就绪”，则任务本身会收到通知，并会尝试轮询其所有三个子future。其中一个将推进，这反过来推进任务的内部状态。

关键是尽量减少任务，尽可能少地完成每项任务。这就是为什么服务器为每个连接生成新任务而不是在与侦听器相同的任务中处理连接的原因。

好吧，实际上有一种方法可以让任务知道哪个子future使用FuturesUnordered触发了通知，但通常正确的做法是生成一个新任务。