Canais no Tokio

Tokio fornece canais assíncronos para coordenação de tarefas: mpsc para filas de trabalho, oneshot para respostas únicas, broadcast para fan-out e watch para sinalização do último valor.

Receita

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

use tokio::sync::mpsc;
 
#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    tx.send("job").await.unwrap();
    println!("{}", rx.recv().await.unwrap());
}

Quando usar isso: Pools de workers, requisição/resposta entre tarefas, broadcast de desligamento e notificações de mudança de configuração.

Exemplo de Trabalho

use tokio::sync::{broadcast, oneshot, watch};
 
#[tokio::main]
async fn main() {
    // oneshot: resposta única
    let (otx, orx) = oneshot::channel();
    tokio::spawn(async move {
        otx.send(42).unwrap();
    });
    println!("oneshot: {}", orx.await.unwrap());
 
    // watch: último valor
    let (wtx, mut wrx) = watch::channel("v1".to_string());
    wtx.send("v2".into()).unwrap();
    wrx.changed().await.unwrap();
    println!("watch: {}", *wrx.borrow());
 
    // broadcast: todos os assinantes
    let (btx, mut brx) = broadcast::channel(16);
    btx.send("event").unwrap();
    println!("broadcast: {}", brx.recv().await.unwrap());
}

O que isso demonstra:

Cada tipo de canal resolve um padrão de coordenação diferente.
watch sobrescreve - apenas o último valor importa.
broadcast pode atrasar se os assinantes forem lentos (erro configurável).

Mergulho Profundo

Como Funciona

mpsc: Multi-produtor, consumidor único; send().await limitado aplica backpressure.
oneshot: Uma mensagem, perfeito para callback estilo RPC entre duas tarefas.
broadcast: Receivers clonados; cada um recebe mensagens; consumidores lentos podem ser descartados.
watch: Sender::send atualiza; Receiver::changed().await espera por um novo valor.

Seletor de Canal

Canal	Mensagens	Consumidores	Uso Típico
mpsc	muitas	um	fila de trabalho
oneshot	uma	um	resposta única
broadcast	muitas	muitas	eventos, ticks
watch	último	muitas	configuração, flag de desligamento

Notas de Rust

use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;
 
let (tx, rx) = mpsc::channel(8);
let mut stream = ReceiverStream::new(rx);
while let Some(item) = stream.next().await {
    println!("{item}");
}

Adapte mpsc::Receiver para Stream para processamento estilo iterador.
Semaphore complementa canais para limitar trabalho em andamento.
Prefira mpsc em vez de Mutex<VecDeque> para limites de tarefas.

Armadilhas

mpsc sem limite - channel(1) com consumidor lento ainda bloqueia na capacidade; mpsc sem limite não está no mpsc padrão do Tokio (sempre limitado). Correção: Dimensionar o buffer para o pico esperado.
oneshot usado duas vezes - O segundo send falha. Correção: Apenas um receiver; clone os dados antes de enviar se necessário em outro lugar.
Erro de atraso no broadcast - Assinante lento perde mensagens. Correção: Aumentar a capacidade ou lidar com RecvError::Lagged.
watch sem changed() - Pode ler dados obsoletos sem notificação. Correção: changed().await antes de confiar em atualizações em loops.
Bloqueio em callbacks de canal - Paralisa o runtime. Correção: Mantenha os manipuladores assíncronos e não bloqueantes.

Alternativas

Alternativa	Usar Quando	Não Usar Quando
`std::sync::mpsc`	Apenas threads síncronas	Tarefas assíncronas aguardando recv
`crossbeam`	Padrões síncronos MPMC	Loops assíncronos puros await
`Mutex + Condvar`	Legado síncrono	Serviços Tokio idiomáticos
Crates de Ator	Máquinas de estado complexas	Requisição/resposta simples

FAQs

tamanho do buffer mpsc?

Corresponda ao tamanho do pico e à tolerância a backpressure - muito pequeno paralisa os produtores; muito grande esconde sobrecarga.

oneshot vs mpsc de 1?

oneshot é mais leve para resposta única; mpsc quando múltiplas mensagens são possíveis.

broadcast vs watch?

broadcast entrega cada evento; watch mantém apenas o último snapshot.

Fechar mpsc?

Descarte todos os handles Sender; recv retorna None.

try_recv?

Consulta não assíncrona - use em select! ou ao conectar código síncrono com cuidado.

fairness?

mpsc FIFO por remetente entrelaçado - não é fairness global estrita entre muitos produtores.

Axum + canais?

Crie uma tarefa worker com receiver mpsc; manipuladores enviam trabalhos via Sender clonado no estado.

recv cancel-safe?

mpsc::recv em loops select! é cancel-safe de acordo com a documentação do Tokio - ainda assim, projete para wakeups perdidos em configurações complexas.

Semaphore vs canal?

Semaphore limita a concorrência; canal também transporta dados - frequentemente usados juntos.

testando?

Use buffers pequenos para forçar backpressure em testes; timeout em recv para pegar travamentos.

Relacionado

Canais - canais síncronos
Streams - ReceiverStream
select! & Concorrência - select de canal
Estado Compartilhado em Async - padrão de ator
Desligamento Gracioso - canais de desligamento

Versões da Stack: 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+.

Canais no Tokio

Receita

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

use tokio::sync::mpsc;
 
#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    tx.send("job").await.unwrap();
    println!("{}", rx.recv().await.unwrap());
}

Quando usar isso: Pools de workers, requisição/resposta entre tarefas, broadcast de desligamento e notificações de mudança de configuração.

Exemplo de Trabalho

use tokio::sync::{broadcast, oneshot, watch};
 
#[tokio::main]
async fn main() {
    // oneshot: resposta única
    let (otx, orx) = oneshot::channel();
    tokio::spawn(async move {
        otx.send(42).unwrap();
    });
    println!("oneshot: {}", orx.await.unwrap());
 
    // watch: último valor
    let (wtx, mut wrx) = watch::channel("v1".to_string());
    wtx.send("v2".into()).unwrap();
    wrx.changed().await.unwrap();
    println!("watch: {}", *wrx.borrow());
 
    // broadcast: todos os assinantes
    let (btx, mut brx) = broadcast::channel(16);
    btx.send("event").unwrap();
    println!("broadcast: {}", brx.recv().await.unwrap());
}

O que isso demonstra:

Cada tipo de canal resolve um padrão de coordenação diferente.
watch sobrescreve - apenas o último valor importa.
broadcast pode atrasar se os assinantes forem lentos (erro configurável).

Mergulho Profundo

Como Funciona

mpsc: Multi-produtor, consumidor único; send().await limitado aplica backpressure.
oneshot: Uma mensagem, perfeito para callback estilo RPC entre duas tarefas.
broadcast: Receivers clonados; cada um recebe mensagens; consumidores lentos podem ser descartados.
watch: Sender::send atualiza; Receiver::changed().await espera por um novo valor.

Seletor de Canal

Canal	Mensagens	Consumidores	Uso Típico
mpsc	muitas	um	fila de trabalho
oneshot	uma	um	resposta única
broadcast	muitas	muitas	eventos, ticks
watch	último	muitas	configuração, flag de desligamento

Notas de Rust

use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;
 
let (tx, rx) = mpsc::channel(8);
let mut stream = ReceiverStream::new(rx);
while let Some(item) = stream.next().await {
    println!("{item}");
}

Adapte mpsc::Receiver para Stream para processamento estilo iterador.
Semaphore complementa canais para limitar trabalho em andamento.
Prefira mpsc em vez de Mutex<VecDeque> para limites de tarefas.

Armadilhas

mpsc sem limite - channel(1) com consumidor lento ainda bloqueia na capacidade; mpsc sem limite não está no mpsc padrão do Tokio (sempre limitado). Correção: Dimensionar o buffer para o pico esperado.
oneshot usado duas vezes - O segundo send falha. Correção: Apenas um receiver; clone os dados antes de enviar se necessário em outro lugar.
Erro de atraso no broadcast - Assinante lento perde mensagens. Correção: Aumentar a capacidade ou lidar com RecvError::Lagged.
watch sem changed() - Pode ler dados obsoletos sem notificação. Correção: changed().await antes de confiar em atualizações em loops.
Bloqueio em callbacks de canal - Paralisa o runtime. Correção: Mantenha os manipuladores assíncronos e não bloqueantes.

Alternativas

Alternativa	Usar Quando	Não Usar Quando
`std::sync::mpsc`	Apenas threads síncronas	Tarefas assíncronas aguardando recv
`crossbeam`	Padrões síncronos MPMC	Loops assíncronos puros await
`Mutex + Condvar`	Legado síncrono	Serviços Tokio idiomáticos
Crates de Ator	Máquinas de estado complexas	Requisição/resposta simples

FAQs

tamanho do buffer mpsc?

Corresponda ao tamanho do pico e à tolerância a backpressure - muito pequeno paralisa os produtores; muito grande esconde sobrecarga.

oneshot vs mpsc de 1?

oneshot é mais leve para resposta única; mpsc quando múltiplas mensagens são possíveis.

broadcast vs watch?

broadcast entrega cada evento; watch mantém apenas o último snapshot.

Fechar mpsc?

Descarte todos os handles Sender; recv retorna None.

try_recv?

Consulta não assíncrona - use em select! ou ao conectar código síncrono com cuidado.

fairness?

mpsc FIFO por remetente entrelaçado - não é fairness global estrita entre muitos produtores.

Axum + canais?

Crie uma tarefa worker com receiver mpsc; manipuladores enviam trabalhos via Sender clonado no estado.

recv cancel-safe?

mpsc::recv em loops select! é cancel-safe de acordo com a documentação do Tokio - ainda assim, projete para wakeups perdidos em configurações complexas.

Semaphore vs canal?

Semaphore limita a concorrência; canal também transporta dados - frequentemente usados juntos.

testando?

Use buffers pequenos para forçar backpressure em testes; timeout em recv para pegar travamentos.

Relacionado

Canais - canais síncronos
Streams - ReceiverStream
select! & Concorrência - select de canal
Estado Compartilhado em Async - padrão de ator
Desligamento Gracioso - canais de desligamento

Versões da Stack: 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+.

Canais no Tokio

Buscar na documentação

Canais no Tokio