Streams

Stream é o primo assíncrono de Iterator: ele produz uma sequência de itens ao longo do tempo, impulsionado por .await em vez de .next(). Use streams para APIs paginadas, WebSockets, blocos de arquivos e feeds de eventos.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

use tokio_stream::{self as stream, StreamExt};
 
#[tokio::main]
async fn main() {
    let mut s = stream::iter(vec![10, 20, 30]);
    while let Some(n) = s.next().await {
        println!("{n}");
    }
}

Quando usar isso: Consumir HTTP paginado, cursores de banco de dados, mensagens Kafka/NATS ou eventos SSE/WebSocket.

Exemplo de Trabalho

use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
async fn produce(tx: mpsc::Sender<i32>) {
    for i in 1..=5 {
        tx.send(i).await.unwrap();
    }
}
 
#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(8);
    tokio::spawn(produce(tx));
 
    let mut stream = ReceiverStream::new(rx);
    let sum: i32 = stream.map(|n| n * 2).sum().await;
    println!("{sum}");
}

O que isso demonstra:

• O receptor do canal se adapta a Stream via ReceiverStream.
• StreamExt fornece map, filter, take, buffer_unordered.
• sum().await consome o stream.

Mergulho Profundo

Como Funciona

• Stream::poll_next: Semelhante a Future::poll, mas retorna Option<Item> até None.
• Backpressure: Consumidores lentos aplicam naturalmente pressão em canais limitados.
• Fusão com futures: Muitos aplicativos mapeiam streams para respostas (Axum SSE, gRPC streaming).
• StreamExt: Métodos ergonômicos dos crates futures / tokio-stream.

Adaptadores Comuns

Notas de Rust

use futures::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
 
struct Counter {
    n: u32,
    max: u32,
}
 
impl Stream for Counter {
    type Item = u32;
 
    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.n >= self.max {
            Poll::Ready(None)
        } else {
            let v = self.n;
            self.n += 1;
            Poll::Ready(Some(v))
        }
    }
}

• Implementações manuais de Stream espelham as regras de pinning de Future.
• Prefira adaptadores em vez de implementações manuais no código da aplicação.
• A macro try_stream! constrói streams falíveis de forma ergonômica.

Armadilhas

• Produção ilimitada - Produtor rápido sobrecarrega a memória sem limites. Correção: mpsc limitado ou stream::iter com limites de taxa.
• Esquecer de impulsionar o stream - Criar um stream sem consumi-lo vaza tarefas. Correção: Sempre drene, cancele ou descarte explicitamente.
• Bloqueio em adaptadores de stream - Cláusulas map não devem bloquear. Correção: then com offload assíncrono.
• Mistura de crates de Stream - Importe StreamExt do crate que corresponde ao seu tipo de stream. Correção: use tokio_stream::StreamExt vs futures::StreamExt (habilite recursos).
• Sem for await em Rust estável - Use while let Some ou StreamExt::next. Correção: Loop com .await em next().

Alternativas

FAQs

Relacionados

• Noções Básicas de Async - introdução à iteração assíncrona
• Combinando Futures - select com streams
• Canais em Tokio - ReceiverStream
• I/O Assíncrono - streams de bytes
• Cancelamento e Timeouts - timeouts de stream

Versões da Pilha: Esta página foi escrita para Rust 1.97.0 (edição 2024), Tokio 1.x, Axum 0.8, serde 1.0, sqlx 0.8, clap 4, e Polars 0.46+.