Async Rust com Tokio: async/await, task e channel

1. Visão Geral

Rust tem nativamente a sintaxe async/await mas não vem com um runtime para executar Future. Tokio é o runtime mais popular para isso. Ele fornece scheduler, timer, I/O assíncrono e ferramentas de sincronização.

Este artigo vai desde a primeira função async até as operações mais comuns: spawn task, executar em paralelo com join!, definir timeout com select!, compartilhar estado entre tasks, enviar dados via channel e lidar com código blocking. A última parte fala sobre como escrever testes e erros comuns.

2. Instalação

Você adiciona Tokio ao Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

A feature full ativa todos os módulos: runtime multi-thread, time, sync, net, macros… Isso é conveniente ao aprender. Em um projeto real, você deve ativar apenas as features que realmente usa para reduzir o tempo de compilação, por exemplo ["rt-multi-thread", "macros", "time", "sync"]. A feature test-util permite controlar o tempo em testes. Vamos usá-lo na seção 8.

3. Primeira Função Async

3.1. #[tokio::main] e .await

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // simular chamada ao banco de dados
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

Explicação:

• async fn fetch_order não retorna uma String imediatamente. Ela retorna uma impl Future<Output = String>.
• tokio::time::sleep é a versão async do sleep. Enquanto aguarda, a thread é devolvida ao runtime para executar outras tarefas.
• #[tokio::main] é uma macro. Ela cria um runtime multi-thread e então executa main dentro daquele runtime usando block_on.

Order #42

3.2. Future é Lazy

Esta é uma diferença importante em relação às Promises em JavaScript: futures em Rust não fazem nada até serem .awaited ou polled.

async fn load_customer(id: u32) -> String {
    println!("Carregando cliente {id}");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // nenhuma linha da função foi executada ainda
    println!("Antes do await");
    let customer = fut.await;
    println!("{customer}");
}

Antes do await
Carregando cliente 7
Customer 7

A linha “Carregando cliente 7” só aparece depois de chamar .await. Se você esquecer .await, a função nunca será executada. O compilador avisará unused implementer of Future that must be used, então não ignore este aviso.

4. Spawn Task

.await executa futures sequencialmente. Para executá-las simultaneamente, você usa tokio::spawn para criar uma task. Uma task é uma “green thread” leve gerenciada pelo Tokio.

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("task sofreu panic");
        println!("{order}");
    }
    println!("Tempo total: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
Tempo total: ~300 ms

Se executadas sequencialmente, o tempo total seria 100 + 200 + 300 = 600 ms. Quando spawn, as três tasks são executadas simultaneamente, então o total é apenas cerca de 300 ms.

Alguns pontos a serem observados:

• tokio::spawn retorna JoinHandle<T>. Quando você .await o handle, você recebe Result<T, JoinError>. O resultado é Err se a task sofrer panic ou for cancelada.
• O future passado deve satisfazer Send + 'static, porque a task pode executar em uma thread diferente e viver além do escopo atual. É por isso que você precisa de async move.
• Uma task começa a executar assim que é spawned, sem esperar pelo .await do handle.

Nota: Fazer drop de um JoinHandle não cancela a task. A task continua executando em modo detached. Para cancelar, você chama handle.abort().

Quando o número de tasks muda, você pode usar tokio::task::JoinSet. Ele permite obter resultados na ordem em que as tasks são concluídas usando join_set.join_next().await.

5. Executar Concorrentemente em uma Única Task: join!, select!, timeout

5.1. join! e try_join!

Nem sempre você precisa spawn. tokio::join! faz poll de vários futures simultaneamente na mesma task e aguarda que todos sejam concluídos:

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

O tempo total é igual ao future mais lento, cerca de 200 ms. Os futures não precisam de 'static, então podem emprestar variáveis locais.

Quando as funções retornam Result, você usa tokio::try_join!. Essa macro para no primeiro erro e descarta os futures restantes:

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

Aqui fetch_order_checked e fetch_customer_checked são funções async que retornam Result<String, AppError>.

5.2. timeout

use tokio::time::{timeout, Duration};

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("Excedeu o tempo de espera"),
}

Excedeu o tempo de espera

fetch_order leva 200 ms, ultrapassando o limite de 50 ms. Nesse caso, timeout retorna Err(Elapsed) e o future interno sofre drop, ou seja, é cancelado.

5.3. select!

select! aguarda o primeiro ramo ser concluído e depois cancela os demais:

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("Processamento concluído {order}"),
        _ = shutdown_rx => println!("Recebido sinal de shutdown, parando processamento"),
    }
}

Recebido sinal de shutdown, parando processamento

O sinal de shutdown chega após 100 ms, mais cedo que fetch_order (200 ms), então o segundo ramo vence.

Nota: Cancelamento em Rust significa que o future é descartado no ponto .await mais próximo. Se o ramo sendo cancelado está no meio de uma operação não atômica, por exemplo, leu metade de uma mensagem do socket, os dados podem ser perdidos. Você deve verificar a seção “cancel safety” na documentação de cada função antes de usá-la em select! dentro de um loop.

6. Compartilhar Dados entre Tasks

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("Restante: {}", inventory.lock().unwrap()["laptop"]);
}

Restante: 5

Aqui usamos std::sync::Mutex, não tokio::sync::Mutex. Esta é a escolha recomendada quando a seção crítica é curta e não há .await dentro, porque o mutex padrão é mais rápido.

Você só precisa de tokio::sync::Mutex quando deve manter um lock através de um ponto .await, por exemplo, ao manter uma conexão compartilhada enquanto aguarda I/O.

6.2. Canal mpsc

Em vez de compartilhar memória, você pode passar mensagens entre tasks via canal:

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // fechar o sender original para que o loop abaixo termine

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("Recebido {order:?}");
        total += order.amount;
    }
    println!("Total: {total}");
}

Recebido Order { id: 1, amount: 100 }
Recebido Order { id: 3, amount: 300 }
Recebido Order { id: 2, amount: 200 }
Total: 600

A ordem de recebimento pode variar entre execuções. O parâmetro 32 é a capacidade do buffer. Quando o buffer está cheio, send aguardará, criando um mecanismo de back-pressure natural. recv() retorna None quando todos os senders foram descartados. Se você esquecer drop(tx), o programa ficará travado no loop while.

Além de mpsc, Tokio também tem:

• oneshot: enviar um único valor.
• broadcast: vários consumers, cada um recebe todas as mensagens.
• watch: consumers só se importam com o valor mais recente.

7. Lidar com Código Blocking

Tokio usa cooperative scheduling. Tasks só cedem a thread nos pontos .await. Uma operação CPU pesada ou I/O síncrono ocupará o worker thread e fará outras tasks ficarem com fome de recursos.

fn hash_password(password: &str) -> String {
    // simular operação intensiva de CPU, por exemplo bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("task de bloqueio sofreu panic");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking move a closure para um thread pool separado destinado a operações de bloqueio. Assim, os worker threads do runtime permanecem livres para lidar com outras requisições.

Nota: Não chame std::thread::sleep, std::fs::read ou cliente HTTP síncrono diretamente dentro de funções async. Use as versões async (tokio::time::sleep, tokio::fs) ou encapsule-as em spawn_blocking. Para computação paralela em grande escala, Rayon geralmente é mais apropriado.

8. Testar Código Async

A macro #[tokio::test] cria um runtime para cada teste. Por padrão, este é um runtime single-thread.

use std::time::Duration;
use tokio::time::{sleep, timeout};

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true (requer a feature test-util) congela o relógio do Tokio. Quando todas as tasks estão aguardando um timer, o runtime pula automaticamente para o próximo marco. Assim, testes sobre timeout são executados quase instantaneamente e fornecem resultados estáveis, independentemente da velocidade da máquina CI.

Se você precisa verificar o comportamento em múltiplas threads, use #[tokio::test(flavor = "multi_thread", worker_threads = 2)].

9. Erros Comuns

Manter std::sync::MutexGuard através de .await. O guard não é Send, então o future também não será Send. Nesse caso tokio::spawn reporta erro de compilação future cannot be sent between threads safely. A correção é estreitar o escopo do guard com um bloco { ... } antes do .await, ou trocar para tokio::sync::Mutex.

Criar runtimes aninhados. Chamar Runtime::new().block_on(...) dentro de código já executando em Tokio causará panic com a mensagem Cannot start a runtime from within a runtime. Em contexto async, você só precisa de .await.

Usar async quando não é necessário. Uma ferramenta CLI que lê alguns arquivos, ou um job de computação pura, não se beneficia do Tokio. Você apenas paga o preço de tempo de compilação e complexidade adicionais. Async faz diferença quando há muitas operações de I/O simultâneas: servidor web, crawler, proxy, cliente chamando múltiplos serviços.

Spawn ilimitado. Tasks são baratas mas não grátis. Spawnar uma task para cada item em uma lista de um milhão de elementos pode esgotar recursos downstream, como o pool de conexões do banco de dados. Limite a concorrência usando tokio::sync::Semaphore ou buffer_unordered do crate futures.

Ignorar JoinError. Se você .await o handle mas usa let _ =, panic dentro da task será engolido. Você deve logar o erro ou propagá-lo para cima.

10. Conclusão

Este artigo cobriu os pontos essenciais de async Rust com Tokio:

• Futures são lazy e só executam quando .awaited.
• tokio::spawn cria tasks que executam concorrentemente. O future passado deve ser Send + 'static.
• join! e try_join! executam múltiplos futures na mesma task. select! e timeout lidam com race e cancelamento.
• Dados compartilhados podem ser compartilhados via Arc<Mutex<T>> ou passados através de canais como mpsc e oneshot.
• Código de bloqueio deve ser colocado em spawn_blocking.
• #[tokio::test(start_paused = true)] ajuda testes com tempo a executar rápido e de forma estável.

Uma vez que você domina essas partes, pode continuar com I/O de rede via tokio::net, ou construir web services no Tokio usando frameworks como Axum.