Los canales std::sync::mpsc implementan el modelo de comunicación de “Múltiples Productores, Un Solo Consumidor” (Multiple Producer, Single Consumer). En lugar de compartir memoria mediante bloqueos (Mutex), este modelo fomenta el paso de mensajes: un hilo “envía” la propiedad de un dato a otro hilo a través de un canal. Esto evita condiciones de carrera y simplifica el razonamiento sobre la concurrencia al seguir la filosofía: “No comuniques compartiendo memoria; comparte memoria comunicando”.

Un canal se compone de dos extremos: un Sender<T> (emisor) y un Receiver<T> (receptor). El Sender puede clonarse para permitir que múltiples hilos actúen como productores. Sin embargo, el Receiver no es clonable, lo que garantiza que solo un hilo sea el encargado de procesar los mensajes.

Existen dos tipos de canales:
1. Canales asíncronos (mpsc::channel): Tienen un buffer ilimitado. El emisor nunca se bloquea al enviar, lo cual es eficiente pero puede agotar la memoria si el productor es mucho más rápido que el consumidor.
2. Canales síncronos (mpsc::sync_channel(n)): Tienen un buffer limitado a n elementos. Si el buffer se llena, el Sender se bloquea hasta que el receptor consuma un elemento, implementando un mecanismo de contrapresión (backpressure).

Si se intenta enviar un mensaje a un canal donde el receptor ha sido destruido, send devolverá un error. Del mismo modo, si el receptor intenta recibir datos pero todos los Sender han sido eliminados (dropped), la operación se cerrará.

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

fn main() {
    // Se crea un canal de comunicación asíncrono.
    // tx: Sender (emisor), rx: Receiver (receptor).
    let (tx, rx) = mpsc::channel();

    // Clonamos el emisor para crear un segundo productor.
    // Esto es lo que permite el "Multiple Producer".
    let tx_clon = tx.clone();

    // Hilo 1: Produce los primeros números.
    thread::spawn(move || {
        let valores = vec![1, 2, 3];
        for val in valores {
            tx_clon.send(val).unwrap(); // Enviamos el valor al canal.
            thread::sleep(Duration::from_millis(500));
        }
        // Al terminar este hilo, tx_clon se destruye.
    });

    // Hilo 2: Produce otros números usando el emisor original.
    thread::spawn(move || {
        let valores = vec![4, 5, 6];
        for val in valores {
            tx.send(val).unwrap(); // Enviamos el valor al canal.
            thread::sleep(Duration::from_millis(100));
        }
        // Al terminar este hilo, tx se destruye.
    });

    // El hilo principal actúa como el único consumidor.
    // El bucle for iterará sobre el receptor hasta que todos los 
    // Sender (tx y tx_clon) hayan sido droppeados.
    for mensaje in rx {
        println!("Consumido: {}", mensaje);
    }

    println!("Canal cerrado. Fin de la ejecución.");
}

Explicación del Código

  • mpsc::channel(): Inicializa el canal devolviendo una tupla que contiene tx (de tipo Sender<i32>) y rx (de tipo Receiver<i32>).
  • tx.clone(): Crea una nueva instancia del emisor. En el ejemplo, tx_clon es una copia de tx que permite que el primer hilo tenga su propio manejador para enviar datos.
  • thread::spawn(move || { ... }): Crea un hilo nuevo. El uso de move es crucial para transferir la propiedad de tx_clon o tx al interior del hilo.
  • tx_clon.send(val).unwrap(): Envía el valor val al canal. Si el canal estuviera cerrado, unwrap() causaría un pánico.
  • rx: El iterador for mensaje in rx es la forma idiomática de consumir un canal. El hilo principal se queda bloqueado en el recv() interno de este iterador hasta que llega un mensaje.
  • Cierre del canal: El bucle for termina automáticamente cuando todos los emisores (tx_clon y tx) salen de su ámbito (scope) o son destruidos. Una vez que no hay más hilos capaces de enviar, rx sabe que no habrá más mensajes y finaliza la iteración.

N° 81