前言#
昨天我们了解了Pin的用法和它的作用
今天接着往下学
Stream trait#
Stream这个trait和Future有些像，但是它可以让步(yield)更多的值，在编译前。
这一点又和Iterator挺像的。
trait Stream {
/// The type of the value yielded by the stream.
type Item;
/// Attempt to resolve the next item in the stream.
/// Returns `Poll::Pending` if not ready, `Poll::Ready(Some(x))` if a value
/// is ready, and `Poll::Ready(None)` if the stream has completed.
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>)
-> Poll<Option<Self::Item>>;
}一般使用Stream的场景是channel的Receiver也就是接收器(不要忘了前提是future，也就是异步场景)，我们day2写的例子用的是循环来实现的， 现在我们可以用迭代来实现，这样从性能/优雅度来说是比循环高的。
来看个例子
async fn send_recv() {
const BUFFER_SIZE: usize = 10;
let (mut tx, mut rx) = mpsc::channel::<i32>(BUFFER_SIZE);
tx.send(1).await.unwrap();
tx.send(2).await.unwrap();
drop(tx);
// `StreamExt::next` is similar to `Iterator::next`, but returns a
// type that implements `Future<Output = Option<T>>`.
assert_eq!(Some(1), rx.next().await);
assert_eq!(Some(2), rx.next().await);
assert_eq!(None, rx.next().await);
}迭代(Iteration)和并发(Concurrency)#
和同步的Iterator类似，Stream可以有很多方法可以迭代和修改数据，比如map、filter、fold等。
当然，还有它们的表兄弟try_map、try_filter、try_fold，会在错误时退出。
不过for循环并不能用于Stream中，你可以用while let搭配next/try_next来实现循环。
来看个使用例子
async fn sum_with_next(mut stream: Pin<&mut dyn Stream<Item = i32>>) -> i32 {
use futures::stream::StreamExt; // for `next`
let mut sum = 0;
while let Some(item) = stream.next().await {
sum += item;
}
sum
}
async fn sum_with_try_next(
mut stream: Pin<&mut dyn Stream<Item = Result<i32, io::Error>>>,
) -> Result<i32, io::Error> {
use futures::stream::TryStreamExt; // for `try_next`
let mut sum = 0;
while let Some(item) = stream.try_next().await? {
sum += item;
}
Ok(sum)
}不过这么做就意味着和并发无关了，毕竟异步就是用来缓解并发的，如果不能用于并发就有些搞笑了。。
所以我们用for_each_concurrent和try_for_each_concurrent来替代上面的循环或者同步的方法。
async fn jump_around(
mut stream: Pin<&mut dyn Stream<Item = Result<u8, io::Error>>>,
) -> Result<(), io::Error> {
use futures::stream::TryStreamExt; // for `try_for_each_concurrent`
const MAX_CONCURRENT_JUMPERS: usize = 100;
stream.try_for_each_concurrent(MAX_CONCURRENT_JUMPERS, |num| async move {
jump_n_times(num).await?;
report_n_jumps(num).await?;
Ok(())
}).await?;
Ok(())
}总结#
这章内容好水，估计是还没完善的。