Canales en Rust: paso de mensajes entre hilos

En la página de concurrencia, los hilos (threads) compartían datos bloqueándolos con Arc<Mutex<T>>. Funciona, pero hay otra forma que muchas veces es más sencilla: en lugar de compartir datos, los hilos se envían mensajes entre sí.

Piensa en un channel (canal) como un tubo con dos extremos. Un hilo mete valores por el extremo del emisor (sender). Otro hilo los saca por el extremo del receptor (receiver), en el mismo orden. Nadie comparte nada: cada valor se mueve de un hilo al otro. La comunidad de Go tiene un dicho que Rust adopta aquí: “No te comuniques compartiendo memoria; comparte memoria comunicándote”.

Tu primer canal

El canal de la biblioteca estándar está en std::sync::mpsc. El nombre significa multiple producer, single consumer (varios productores, un solo consumidor): muchos emisores, un receptor.

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

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

    thread::spawn(move || {
        let message = String::from("hola desde el hilo trabajador");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main recibió: {received}");
}

Salida:

main recibió: hola desde el hilo trabajador

▶ Pruébalo tú mismo

Paso a paso:

• mpsc::channel() devuelve una pareja: un Sender y un Receiver. Por tradición se llaman tx (transmitir) y rx (recibir).
• move le entrega tx al hilo nuevo, igual que movías datos a los hilos en la página de concurrencia.
• tx.send(valor) mete un valor en el canal. Devuelve un Result, porque falla si el receptor ya no existe.
• rx.recv() espera hasta que llega un valor. Solo devuelve Err cuando ya no queda ningún emisor y el canal está vacío.

Fíjate en que no necesitamos join(). recv() ya espera al mensaje, así que main no puede terminar antes de tiempo.

Recibir muchos valores

Un Receiver funciona como un iterador. Un bucle for sobre él llama a recv() por ti una y otra vez, y el bucle termina solo cuando todos los emisores se han liberado (drop):

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

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

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx se libera aquí, cuando termina el hilo
    });

    for value in rx {
        println!("recibido {value}");
    }
    println!("canal cerrado, todo listo");
}

Salida:

recibido 10
recibido 20
recibido 30
canal cerrado, todo listo

▶ Pruébalo tú mismo

Aquí solo hay un emisor, así que los valores llegan exactamente en el orden en que se enviaron. Esta regla de “el bucle se detiene cuando ya no quedan emisores” es importante. Tenla presente para la siguiente sección.

Varios productores: clona el emisor

Para que varios hilos puedan enviar, dale a cada uno su propia copia del emisor con tx.clone(). Todas las copias alimentan el mismo canal:

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

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

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("el trabajador {id} se reporta")).unwrap();
        });
    }
    drop(tx); // libera el original, o el bucle de abajo nunca termina

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Salida:

el trabajador 1 se reporta
el trabajador 2 se reporta
el trabajador 3 se reporta

▶ Pruébalo tú mismo

Dos cosas en las que fijarte:

• drop(tx) importa. Cada hilo libera su clon al terminar, pero main todavía tiene el original. Mientras exista un emisor, el receptor sigue esperando, y el programa se quedaría colgado para siempre. Liberarlo equivale a decir “main no va a enviar nada”.
• Los hilos terminan en cualquier orden. Los mensajes pueden llegar como trabajador 2, trabajador 1, trabajador 3. Los recogemos y los ordenamos para que la salida impresa sea siempre la misma. Quita el sort() y ejecútalo varias veces para ver cómo cambia el orden.

No esperar para siempre: try_recv y recv_timeout

recv() se bloquea hasta que llega algo. A veces no quieres esperar. Te ayudan dos parientes suyos:

• try_recv() vuelve de inmediato: con un valor, con Empty (todavía nada) o con Disconnected (ya no quedan emisores).
• recv_timeout(duration) espera, pero solo hasta un tiempo límite. Si no llega nada a tiempo, obtienes Timeout.

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

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

    // Todavía no se envió nada: try_recv vuelve enseguida.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Hay un valor esperando, así que lo devuelve sin esperar.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // El emisor ya no existe y el canal está vacío.
    println!("{:?}", rx.try_recv());
}

Salida:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Pruébalo tú mismo

Hacemos join() del hilo antes de recibir, así que el valor ya está esperando y la salida siempre es la misma.

try_recv() es práctico en un bucle que tiene más trabajo que hacer, como el bucle de un juego que revisa si hay mensajes nuevos una vez por fotograma. recv_timeout() sirve para “me rindo si el trabajador tarda demasiado”.

Canales con límite: sync_channel

Un channel() normal no tiene límite de tamaño. Si el emisor es mucho más rápido que el receptor, los mensajes se acumulan en memoria. mpsc::sync_channel(n) crea un canal que guarda como máximo n mensajes. Cuando está lleno, send() espera hasta que el receptor saque uno. A esto se le llama backpressure (contrapresión): un consumidor lento frena automáticamente a un productor rápido.

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

fn main() {
    // Espacio para solo 2 mensajes a la vez.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // se bloquea mientras el búfer está lleno
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("total = {total}");
}

Salida:

total = 15

▶ Pruébalo tú mismo

El productor nunca puede ir más de dos números por delante de main. Con sync_channel(0) no hay búfer en absoluto: cada send() espera hasta que el receptor esté listo para tomar el valor, como cuando entregas un paquete en mano.

Un pequeño pool de trabajadores

Ahora juntemos todo. Un pool de trabajadores (worker pool) es un grupo de hilos que toman tareas de una cola compartida y devuelven resultados. Usa dos canales: uno para las tareas y otro para los resultados.

Hay una trampa: un Receiver no se puede clonar (es de un solo consumidor). Así que los trabajadores comparten el receptor de tareas detrás de un Arc<Mutex<_>>, y se turnan para sacar tareas:

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

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Los receptores no se pueden clonar, así que los trabajadores comparten uno detrás de Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // no hay más tareas
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // les dice a los trabajadores que no hay más trabajo

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} al cuadrado es {square}");
    }
}

Salida:

1 al cuadrado es 1
2 al cuadrado es 4
3 al cuadrado es 9
4 al cuadrado es 16
5 al cuadrado es 25
6 al cuadrado es 36

▶ Pruébalo tú mismo

Cómo se apaga limpiamente:

1. drop(job_tx) cierra el canal de tareas. Entonces el recv() de cada trabajador devuelve Err, y el trabajador sale de su bucle.
2. Cuando un trabajador termina, se libera su clon de result_tx. Ya liberamos el original, así que en cuanto terminan los tres trabajadores, el canal de resultados se cierra.
3. Entonces result_rx.iter() se detiene, y ordenamos los resultados porque los trabajadores terminan en cualquier orden.

Un detalle: job_rx.lock().unwrap().recv() está escrito en una sola sentencia let a propósito. El bloqueo se libera al final de esa sentencia, así que los demás trabajadores pueden tomar la siguiente tarea mientras este todavía está elevando su número al cuadrado.

Canales frente a Arc<Mutex<T>>

Los dos son seguros. Simplemente encajan con formas distintas de problema:

Usa un canal cuando…	Usa Arc<Mutex<T>> cuando…
El trabajo fluye en una dirección: los productores crean cosas y un consumidor las procesa	Muchos hilos necesitan leer y actualizar el mismo valor (un contador, una caché)
Quieres una cola de tareas, eventos o líneas de log	Los datos son grandes y copiarlos o moverlos de un lado a otro sería un desperdicio
Cada valor tiene un solo dueño a la vez	Los hilos necesitan ver el estado actual, no un flujo de actualizaciones

Una buena regla práctica: si puedes describir tu programa como “estos hilos le cuentan algo a ese hilo”, usa un canal. Si es “estos hilos editan todos esta cosa”, usa un Mutex. Muchos programas reales usan los dos, como nuestro pool de trabajadores.

Errores comunes

Usar un valor después de enviarlo. send() toma el ownership (propiedad), igual que cuando pasas un valor a una función. Después de enviarlo, el valor pertenece al receptor:

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

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

    thread::spawn(move || {
        let message = String::from("hola");
        tx.send(message).unwrap();
        println!("enviado {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:28
   |
 8 |         let message = String::from("hola");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("enviado {message}");
   |                            ^^^^^^^ value borrowed here after move

El compilador dice que message se usa después de haberse movido. Es Rust protegiéndote: si el emisor todavía pudiera cambiar message, los dos hilos tocarían los mismos datos. Imprime antes de enviar, o envía un .clone().

Mover el emisor dentro de un bucle de hilos. El primer hilo se queda con tx, así que no queda nada para el segundo:

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

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

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

El mensaje dice que tx ya se movió al closure en una vuelta anterior del bucle. La solución es poner let tx = tx.clone(); al principio del bucle, como en el ejemplo de varios productores.

Un programa que nunca termina. Este no da ningún error de compilación, y eso lo hace traicionero. Si recorres un receptor (for x in rx) mientras algún emisor sigue vivo, a menudo el tx original de main, el bucle espera para siempre. Cuando tu programa “simplemente se cuelga”, comprueba que todos los emisores se liberen.

Siguientes pasos

std::sync::mpsc cubre mucho, pero te encontrarás con dos alternativas populares:

• crossbeam-channel (en inglés) añade canales con varios consumidores (receptores que sí puedes clonar) y una macro select! para esperar en varios canales a la vez.
• tokio::sync::mpsc es el canal para código async. Si usas async/.await (consulta Async/Await), úsalo en lugar del de la biblioteca estándar, porque su recv().await no bloquea todo el hilo.

Resumen

• mpsc::channel() te da una pareja (Sender, Receiver). send() mete un valor y recv() espera al siguiente.
• Clona el emisor para tener varios productores. Siempre hay un único receptor.
• Recorrer un receptor termina cuando todos los emisores se han liberado, así que acuérdate de hacer drop() del original.
• try_recv() y recv_timeout() te permiten comprobar sin esperar para siempre. sync_channel(n) limita la cola a n mensajes.
• Los mensajes de varios hilos llegan en cualquier orden: ordénalos o súmalos si necesitas una salida estable.
• Usa canales cuando los datos fluyen de un hilo a otro. Usa Arc<Mutex<T>> cuando los hilos comparten y editan un mismo valor.

Comprueba lo que aprendiste

Comprueba lo aprendido

1. ¿Qué significa "mpsc"?

Many processes, shared cache (muchos procesos, caché compartida) Message passing, safe concurrency (paso de mensajes, concurrencia segura) Multiple producer, single consumer (varios productores, un solo consumidor) Mutex protected shared channel (canal compartido protegido por un mutex)

2. ¿Cómo haces que tres hilos envíen al mismo canal?

Dándole a cada hilo un tx.clone() Envolviendo el receptor en Rc Llamando tres veces a mpsc::channel() y juntándolos No se puede: un canal solo tiene un emisor

3. Un bucle for value in rx nunca termina. ¿Cuál es la causa más probable?

El canal está lleno Un emisor (a menudo el tx original de main) nunca se liberó Se llamó a recv() demasiado pronto Los valores no eran Copy

4. ¿Qué hace send() cuando un sync_channel(2) ya tiene 2 mensajes?

Descarta el mensaje más antiguo Devuelve un error al instante Provoca un panic Espera hasta que el receptor saque un mensaje

Puntuación: 0 / 4

Para leer más

• Usar el paso de mensajes para transferir datos entre hilos — El libro de Rust (en inglés): la introducción oficial a los canales, con más ejemplos.
• std::sync::mpsc — documentación de la biblioteca estándar (en inglés): todos los métodos de Sender, SyncSender y Receiver.
• Channels — Rust by Example (en inglés): un ejemplo corto y ejecutable con varios hilos productores.
• De un servidor de un solo hilo a uno multihilo — El libro de Rust (en inglés): construye paso a paso un pool de hilos completo con canales y Arc<Mutex<Receiver>>.
• tokio::sync::mpsc — docs.rs (en inglés): el canal async que usarás con Tokio, y en qué se diferencia del de la biblioteca estándar.