Concorrência segura em Rust com Mutex e canal

1. Visão geral

Rust chama sua capacidade de “fearless concurrency”. O compilador bloqueia data race no momento do build, antes que o programa sequer comece a executar. Este artigo apresenta duas ferramentas principais da biblioteca padrão: Mutex (junto com Arc) para compartilhar estado, e o canal mpsc para transmitir mensagens entre threads. Após a leitura, você saberá como usar cada um, escrever testes para eles e evitar armadilhas comuns como deadlock ou manter um lock por muito tempo.

Todos os exemplos usam apenas a biblioteca padrão, sem dependências externas. Você pode criar um projeto com cargo new orders-demo e depois colar o código em src/main.rs.

2. Problema: compartilhando dados entre threads

Suponha que você tenha vários threads processando pedidos e queira contar o número total de pedidos. A forma ingênua de escrever seria assim:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

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

Este código não compila:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

O compilador apresenta dois motivos:

• thread::spawn exige que a closure seja 'static. A thread filha pode viver mais do que main, portanto não pode emprestar variáveis locais.
• Mesmo com move, cada thread receberá sua própria cópia de total_orders. Ninguém está realmente acumulando em uma variável compartilhada.

Para resolver, você precisa de duas coisas: compartilhar ownership com Arc e sincronizar o acesso com Mutex.

3. Compartilhando estado com Arc<Mutex<T>>

3.1. Contando pedidos a partir de vários threads

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

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

    println!("Total de pedidos: {}", *total_orders.lock().unwrap());
}

Total de pedidos: 4000

Explicação de cada parte:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted) permite que vários threads possuam um valor simultaneamente. Mutex garante que em cada momento apenas uma thread possa acessar o valor interno.
• Arc::clone(&total_orders) apenas incrementa o contador de referências, não copia os dados. Cada thread recebe seu próprio Arc via move.
• counter.lock() bloqueia a thread atual até conseguir obter o lock. A função retorna Result<MutexGuard<u32>, PoisonError<…>>.
• MutexGuard é dereferenciado para &mut u32, então *... += 1 modifica o valor diretamente. A guard é descartada ao final da instrução, e o lock é liberado automaticamente.

Você não pode “esquecer de fazer unlock” em Rust. Também não pode ler um valor ignorando o lock, porque os dados estão dentro do Mutex.

Nota: Você não pode substituir Arc por Rc. Rc não implementa Send, e o compilador reportará um erro assim que você tentar passá-lo para thread::spawn. Da mesma forma, você não pode substituir Mutex por RefCell, porque RefCell não é Sync.

3.2. Escopo da MutexGuard

O lock é mantido até a guard ser descartada. Se você atribuir a guard a uma variável, o lock permanece até o final do escopo dessa variável:

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // a guard é descartada aqui, o lock é liberado

    if reserved {
        // Simula uma chamada lenta ao PaymentService, SEM manter o lock
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Cliente {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

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

Cliente 0: true
Cliente 1: true
Cliente 2: false

A ordem das linhas pode variar entre as execuções. O resultado sempre será exatamente dois true e um false.

O bloco { ... } limita o tempo em que o lock é mantido ao mínimo. A parte lenta (chamada de pagamento) é executada depois que o lock foi liberado, então outras threads não precisam esperar. Se precisar liberar o lock mais cedo, você pode chamar drop(stock) explicitamente.

3.3. Mutex envenenado

Se uma thread fizer panic enquanto está mantendo um lock, Rust marca o Mutex como envenenado. O motivo é que os dados podem estar em um estado incompleto:

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService falhou no meio");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Saldo: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex foi envenenado, saldo atual: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService falhou no meio
Mutex foi envenenado, saldo atual: 70

Portanto, .unwrap() após lock() essencialmente significa “se outra thread sofreu panic enquanto mantinha o lock, então panic também”. Para a maioria das aplicações, essa é uma escolha razoável. Se você souber como recuperar os dados, into_inner() retorna a guard para você verificar e corrigir.

4. Transmitindo mensagens com canal

Mutex responde à pergunta “vários threads modificando um dado”. Canal segue a direção oposta: cada dado tem apenas um proprietário em um dado momento, e a ownership é transferida através do canal. Em muitos problemas, essa abordagem é mais simples e menos propensa a erros.

4.1. Múltiplos produtores, um consumidor

std::sync::mpsc é a abreviação de multiple producer, single consumer (múltiplos produtores, um consumidor):

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

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

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

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("cliente-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // importante: libera o sender original

    let mut revenue = 0;
    for order in rx {
        println!("Pedido recebido #{} de {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Receita: {revenue}");
}

Pedido recebido #100 de cliente-1
Pedido recebido #300 de cliente-3
Pedido recebido #200 de cliente-2
Receita: 900000

Explicação:

• mpsc::channel() retorna um par (Sender, Receiver). Cada produtor recebe um clone de tx.
• tx.send(order) move order para o canal. Após essa linha, a thread que enviou não pode mais usar order, e o compilador reportará um erro se você tentar. Assim, duas threads nunca modificam um Order simultaneamente.
• for order in rx itera até que todos os Sender sejam descartados. Se você esquecer a linha drop(tx), o sender original ainda está vivo em main, e o loop vai ficar esperando para sempre.

A ordem de recebimento depende da programação de threads. A receita total sempre será 900000.

4.2. Canal com limite e backpressure

mpsc::channel() tem um buffer sem limite. Se o produtor for mais rápido que o consumidor, a memória crescerá gradualmente. sync_channel(n) limita o buffer a n elementos, e send bloqueará quando o buffer estiver cheio:

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

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Pedido {id} enviado");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Processando pedido {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Pedido 1 enviado
Pedido 2 enviado
Processando pedido 1
Pedido 3 enviado
Processando pedido 2
...

O produtor envia dois pedidos e depois tem que esperar o consumidor remover alguns. Este é um mecanismo natural de backpressure, muito útil para pipelines de processamento de dados. Com sync_channel(0), cada send tem que esperar até que alguém faça um recv (rendez-vous).

5. Quando não precisar de Arc: thread com escopo

Se os threads certamente terminarem antes que a função retorne, thread::scope permite que você empreste dados diretamente, sem precisar de Arc:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // todos os threads no escopo foram feitos join aqui

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

Graças ao scope, o compilador sabe que os threads não vivem mais do que processed e orders. Portanto, emprestar com & é válido. Após o escopo, into_inner() extrai os dados de dentro do Mutex sem precisar de lock.

6. Escrevendo testes para código concorrente

Você deve separar a lógica em funções puras para facilitar o teste. Coloque o código abaixo em src/lib.rs:

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

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

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

O teste empty_payments_returns_zero verifica um caso extremo: se esquecer o drop(tx), esse teste travará indefinidamente em vez de retornar 0. Esse é um indicador claro de um erro.

Há um ponto sutil em count_orders: o valor é atribuído à variável result antes de retornar. O motivo é que uma guard temporária na expressão final da função poderia viver mais do que total, e o compilador reportaria um erro de borrow.

7. Erros comuns

Deadlock por lock em ordem invertida. Thread A bloqueia orders e depois customers, enquanto thread B bloqueia customers e depois orders. Ambas ficarão esperando uma pela outra para sempre. Rust apenas previne data race, não previne deadlock. Você deve estabelecer uma ordem fixa de locks ou mesclar os dois dados em um único Mutex<(A, B)>.

Guard temporária vive mais do que você pensa. Veja o código abaixo:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // o lock ainda está sendo mantido aqui!
}

A temporária na expressão de while let (e match) vive até o final do corpo do loop. Qualquer outra thread tentando empurrar um pedido para queue será bloqueada enquanto process está rodando. A correção é separar: let next = queue.lock().unwrap().pop(); e depois processar.

Chamar lock() novamente na mesma thread. std::sync::Mutex não é reentrante. Chamar lock() duas vezes na mesma thread resultará em deadlock ou panic.

Manter lock enquanto aguarda I/O ou recv(). Qualquer operação bloqueante enquanto mantém a guard torna Mutex um gargalo. Você deve copiar ou clonar os dados necessários, liberar o lock e depois fazer o trabalho lento.

Esquecer drop em Sender. Como visto acima, o consumidor esperará indefinidamente.

Nota: Em código async (Tokio), não mantenha std::sync::MutexGuard através de .await. Use tokio::sync::Mutex ou os canais do Tokio. Além disso, para um inteiro simples como um contador, AtomicU32 é muito mais rápido que Mutex<u32>.

Quando escolher qual?

• Use Mutex quando várias threads precisam ler/modificar o mesmo estado por um longo tempo, como cache ou inventário. Se houver muito mais leitura que escrita, considere RwLock.
• Use canal quando dados fluem de um lugar para outro, como fila de pedidos, pipeline ou pool de workers. Canais fazem com que cada valor tenha apenas um proprietário, tornando a lógica mais fácil de racionalizar.

8. Conclusão

Rust transforma erros concorrentes comuns em erros de compilação. Você não pode compartilhar dados sem Arc, não pode modificar dados sem Mutex, e não pode reusar um valor que foi enviado através de um canal. Arc<Mutex<T>> é adequado para estado compartilhado, enquanto o canal mpsc é adequado para transferir ownership entre threads. sync_channel adiciona backpressure, e thread::scope ajuda a evitar Arc quando o tempo de vida dos threads é claro. O que o compilador não pode proteger é deadlock, guard vivendo muito tempo e sender não sendo descartado. Esses são os lugares onde você precisa revisar com cuidado e escrever testes meticulosamente.