Equipe Rust Brasil 7 min de leitura

Concorrência em Rust: Threads e Async | Rust Brasil

Tutorial de concorrência em Rust: threads, canais mpsc, Mutex, Arc, Send/Sync e introdução ao async/await com Tokio.

rust concorrência threads async tutorial

Introdução

Uma das grandes promessas do Rust é a concorrência sem medo (fearless concurrency). O sistema de ownership e types do Rust previne data races em tempo de compilação, algo que nenhuma outra linguagem de sistemas consegue garantir. Neste tutorial, vamos explorar os mecanismos de concorrência disponíveis no Rust, desde threads básicas até programação assíncrona com Tokio.

Threads com std::thread

A forma mais básica de concorrência em Rust é a criação de threads do sistema operacional usando std::thread::spawn.

Criando Threads

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("Thread filha: contagem {}", i);
            thread::sleep(Duration::from_millis(200));
        }
    });

    for i in 1..=3 {
        println!("Thread principal: contagem {}", i);
        thread::sleep(Duration::from_millis(300));
    }

    // Aguarda a thread filha terminar
    handle.join().expect("A thread filha entrou em pânico");
    println!("Todas as threads concluíram!");
}

Movendo Dados para Threads

Quando precisamos usar dados da thread principal dentro de uma thread filha, usamos a palavra-chave move:

use std::thread;

fn main() {
    let nomes = vec![
        String::from("Alice"),
        String::from("Bruno"),
        String::from("Carla"),
    ];

    let handle = thread::spawn(move || {
        for nome in &nomes {
            println!("Olá, {}!", nome);
        }
        nomes // podemos retornar o valor de volta
    });

    // nomes não está mais disponível aqui - foi movido para a thread
    let nomes_retornados = handle.join().unwrap();
    println!("Nomes processados: {:?}", nomes_retornados);
}

Criando Múltiplas Threads

use std::thread;

fn main() {
    let mut handles = vec![];

    for id in 0..5 {
        let handle = thread::spawn(move || {
            let resultado = calcular(id);
            println!("Thread {}: resultado = {}", id, resultado);
            resultado
        });
        handles.push(handle);
    }

    let total: u64 = handles
        .into_iter()
        .map(|h| h.join().unwrap())
        .sum();

    println!("Soma total: {}", total);
}

fn calcular(n: u32) -> u64 {
    (1..=n as u64 * 1000).sum()
}

Canais com mpsc

Canais (channels) implementam o padrão de passagem de mensagens. O módulo std::sync::mpsc oferece canais com múltiplos produtores e um único consumidor (multiple producer, single consumer).

Canal Básico

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let mensagens = vec![
            String::from("Olá"),
            String::from("da"),
            String::from("thread"),
            String::from("filha"),
        ];

        for msg in mensagens {
            tx.send(msg).expect("Falha ao enviar mensagem");
            thread::sleep(std::time::Duration::from_millis(200));
        }
    });

    // Recebendo mensagens - rx.recv() bloqueia até receber
    for recebida in rx {
        println!("Recebido: {}", recebida);
    }
}

Múltiplos Produtores

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

    // Criando múltiplos produtores clonando o transmissor
    for id in 0..3 {
        let tx_clone = tx.clone();
        thread::spawn(move || {
            for i in 0..3 {
                let msg = format!("Produtor {}: mensagem {}", id, i);
                tx_clone.send(msg).unwrap();
                thread::sleep(Duration::from_millis(100));
            }
        });
    }

    // Importante: dropar o transmissor original para que o loop termine
    drop(tx);

    for msg in rx {
        println!("{}", msg);
    }

    println!("Todos os produtores terminaram.");
}

Canal com Tipo Enum para Mensagens Estruturadas

use std::sync::mpsc;
use std::thread;

enum Tarefa {
    Processar(String),
    Calcular(i32, i32),
    Encerrar,
}

fn main() {
    let (tx, rx) = mpsc::channel();

    let worker = thread::spawn(move || {
        loop {
            match rx.recv() {
                Ok(Tarefa::Processar(texto)) => {
                    println!("Processando: {}", texto.to_uppercase());
                }
                Ok(Tarefa::Calcular(a, b)) => {
                    println!("Resultado: {} + {} = {}", a, b, a + b);
                }
                Ok(Tarefa::Encerrar) => {
                    println!("Worker encerrando...");
                    break;
                }
                Err(_) => {
                    println!("Canal fechado, encerrando worker.");
                    break;
                }
            }
        }
    });

    tx.send(Tarefa::Processar(String::from("rust brasil"))).unwrap();
    tx.send(Tarefa::Calcular(42, 58)).unwrap();
    tx.send(Tarefa::Processar(String::from("concorrência"))).unwrap();
    tx.send(Tarefa::Encerrar).unwrap();

    worker.join().unwrap();
}

Estado Compartilhado: Mutex e Arc

Quando múltiplas threads precisam acessar o mesmo dado, usamos Mutex para exclusão mútua e Arc (Atomic Reference Counting) para compartilhamento seguro entre threads.

Mutex Básico

use std::sync::Mutex;

fn main() {
    let contador = Mutex::new(0);

    {
        let mut num = contador.lock().unwrap();
        *num += 1;
    } // O lock é liberado automaticamente aqui (drop)

    println!("Contador: {}", *contador.lock().unwrap());
}

Arc + Mutex para Múltiplas Threads

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let contador = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let contador = Arc::clone(&contador);
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                let mut num = contador.lock().unwrap();
                *num += 1;
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Resultado final: {}", *contador.lock().unwrap());
    // Sempre será 1000 - sem data races!
}

Exemplo Prático: Cache Compartilhado

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

type Cache = Arc<Mutex<HashMap<String, String>>>;

fn buscar_dado(cache: &Cache, chave: &str) -> String {
    // Tenta ler do cache
    {
        let cache_guard = cache.lock().unwrap();
        if let Some(valor) = cache_guard.get(chave) {
            return valor.clone();
        }
    } // Lock liberado aqui

    // Simula busca lenta
    let valor = format!("valor_para_{}", chave);

    // Armazena no cache
    {
        let mut cache_guard = cache.lock().unwrap();
        cache_guard.insert(chave.to_string(), valor.clone());
    }

    valor
}

fn main() {
    let cache: Cache = Arc::new(Mutex::new(HashMap::new()));
    let mut handles = vec![];

    let chaves = vec!["usuario", "config", "dados", "usuario", "config"];

    for chave in chaves {
        let cache = Arc::clone(&cache);
        let chave = chave.to_string();
        let handle = thread::spawn(move || {
            let resultado = buscar_dado(&cache, &chave);
            println!("Thread obteve: {} -> {}", chave, resultado);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    let cache_final = cache.lock().unwrap();
    println!("\nCache final: {:?}", *cache_final);
}

Traits Send e Sync

O Rust garante segurança em concorrência através de duas traits marker:

  • Send: indica que um tipo pode ser transferido entre threads. Quase todos os tipos em Rust são Send. Uma exceção notável é Rc<T>.
  • Sync: indica que um tipo pode ser referenciado a partir de múltiplas threads. Um tipo T é Sync se &T for Send. Mutex<T> é Sync (se T for Send), enquanto RefCell<T> não é.
use std::sync::{Arc, Mutex};
use std::thread;

// Esta struct é Send + Sync automaticamente porque todos os campos são Send + Sync
struct DadosCompartilhados {
    nome: String,
    valor: i32,
}

fn main() {
    // Arc<Mutex<T>> é Send + Sync, permitindo compartilhamento seguro
    let dados = Arc::new(Mutex::new(DadosCompartilhados {
        nome: String::from("teste"),
        valor: 0,
    }));

    let dados_clone = Arc::clone(&dados);
    let handle = thread::spawn(move || {
        let mut d = dados_clone.lock().unwrap();
        d.valor += 1;
        d.nome = String::from("modificado na thread");
    });

    handle.join().unwrap();

    let d = dados.lock().unwrap();
    println!("{}: {}", d.nome, d.valor);
}

Introdução ao Async/Await com Tokio

Para operações de I/O intensivas (servidores web, acesso a banco de dados, requisições HTTP), o modelo async/await é mais eficiente que threads do SO, pois não cria uma thread para cada tarefa.

Configurando o Tokio

No Cargo.toml:

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

Exemplo Básico com Tokio

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

#[tokio::main]
async fn main() {
    println!("Iniciando tarefas assíncronas...");

    // Executando tarefas concorrentemente
    let (r1, r2, r3) = tokio::join!(
        tarefa_async("A", 2),
        tarefa_async("B", 1),
        tarefa_async("C", 3),
    );

    println!("Resultados: {}, {}, {}", r1, r2, r3);
}

async fn tarefa_async(nome: &str, segundos: u64) -> String {
    println!("Tarefa {} iniciada", nome);
    sleep(Duration::from_secs(segundos)).await;
    println!("Tarefa {} concluída após {}s", nome, segundos);
    format!("resultado_{}", nome)
}

Spawn de Tarefas Assíncronas

use tokio::sync::mpsc;

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

    // Spawn de múltiplas tarefas assíncronas
    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let resultado = formato_async(i).await;
            tx.send(resultado).await.expect("Falha ao enviar");
        });
    }

    // Dropar o transmissor original
    drop(tx);

    // Recebendo resultados
    while let Some(msg) = rx.recv().await {
        println!("Recebido: {}", msg);
    }
}

async fn formato_async(id: u32) -> String {
    tokio::time::sleep(tokio::time::Duration::from_millis(100 * id as u64)).await;
    format!("Tarefa {} processada", id)
}

Async com Tratamento de Erros

use std::time::Duration;

#[tokio::main]
async fn main() {
    match buscar_com_timeout("https://httpbin.org/delay/1").await {
        Ok(corpo) => println!("Sucesso: {} bytes recebidos", corpo.len()),
        Err(e) => eprintln!("Erro: {}", e),
    }
}

async fn buscar_com_timeout(url: &str) -> Result<String, Box<dyn std::error::Error>> {
    let cliente = reqwest::Client::new();
    let resposta = tokio::time::timeout(
        Duration::from_secs(5),
        cliente.get(url).send(),
    )
    .await
    .map_err(|_| "Timeout na requisição")?
    .map_err(|e| format!("Erro HTTP: {}", e))?;

    let corpo = resposta.text().await?;
    Ok(corpo)
}

Comparação: Threads vs Async

AspectoThreads (std::thread)Async (Tokio)
Uso idealCPU-boundI/O-bound
Custo por tarefa~8KB+ de stack~poucos bytes
EscalabilidadeCentenas de threadsMilhões de tarefas
ComplexidadeSimplesRequer runtime
PreempçãoSim (pelo SO)Cooperativa

Conclusão

O Rust oferece um ecossistema robusto para programação concorrente:

  • Threads para paralelismo CPU-bound com garantias de segurança
  • Canais para comunicação segura entre threads via passagem de mensagens
  • Mutex/Arc para estado compartilhado quando necessário
  • Send/Sync como garantias em tempo de compilação contra data races
  • Async/await com Tokio para I/O concorrente de alta performance

A combinação do sistema de ownership com essas primitivas de concorrência torna o Rust único: você obtém performance de linguagem de sistemas com segurança de memória garantida pelo compilador. Pratique com os exemplos deste tutorial e experimente construir seus próprios programas concorrentes!