Async Rust con Tokio: async/await, task y channel

1. Descripción general

Rust tiene sintaxis async/await de serie, pero no incluye un runtime para ejecutar Future. Tokio es el runtime más popular para esto. Proporciona scheduler, timer, I/O asincrónico y herramientas de sincronización.

Este artículo comienza con tu primera función async y llega hasta las operaciones más comunes: spawn task, ejecución paralela con join!, timeout con select!, compartir estado entre task, enviar datos por channel y manejar código blocking. La última parte cubre cómo escribir tests y errores frecuentes.

2. Instalación

Añade Tokio a tu Cargo.toml:

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

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

La feature full activa todos los módulos: runtime multihilo, time, sync, net, macros… Es conveniente cuando estás aprendiendo. En proyectos reales, solo deberías activar las features que realmente usas para reducir el tiempo de compilación, por ejemplo ["rt-multi-thread", "macros", "time", "sync"]. La feature test-util permite controlar el tiempo en los tests. La usaremos en la sección 8.

3. Tu primera función async

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

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // simula una consulta a BD
    format!("Order #{id}")
}

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

Explicación:

• async fn fetch_order no devuelve un String inmediatamente. Devuelve un impl Future<Output = String>.
• tokio::time::sleep es la versión asincrónica de sleep. Mientras espera, el thread se devuelve al runtime para ejecutar otra cosa.
• #[tokio::main] es una macro. Crea un runtime multihilo y ejecuta main dentro de ese runtime usando block_on.

Order #42

3.2. Future es lazy

Esta es una diferencia grande con las Promise en JavaScript: los future en Rust no hacen nada hasta que se les hace .await o se les hace poll.

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

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // aún no se ejecuta nada dentro de la función
    println!("Antes de await");
    let customer = fut.await;
    println!("{customer}");
}

Antes de await
Cargando cliente 7
Cliente 7

La línea “Cargando cliente 7” solo aparece después de llamar a .await. Si olvidas .await, la función nunca se ejecutará. El compilador te advertirá con unused implementer of Future that must be used, así que no ignores esta advertencia.

4. Spawn task

.await ejecuta los future secuencialmente. Para que se ejecuten concurrentemente, usas tokio::spawn para crear un task. Un task es un “green thread” ligero gestionado por 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 entró en pánico");
        println!("{order}");
    }
    println!("Tiempo total: ~{} ms", start.elapsed().as_millis());
}

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

Si lo ejecutaras secuencialmente, el tiempo total sería 100 + 200 + 300 = 600 ms. Con spawn, los tres task se ejecutan concurrentemente, así que el total es aproximadamente 300 ms.

Algunos puntos importantes:

• tokio::spawn devuelve un JoinHandle<T>. Cuando haces .await al handle, obtienes un Result<T, JoinError>. El resultado es Err si el task entró en pánico o fue cancelado.
• El future que pases debe cumplir Send + 'static, porque el task puede ejecutarse en otro thread y vivir más allá del scope actual. Por eso necesitas async move.
• El task comienza a ejecutarse inmediatamente después de ser spawn, no espera a que hagas .await al handle.

Nota: Descartar un JoinHandle no cancela el task. El task sigue ejecutándose en modo detached. Para cancelarlo, llama a handle.abort().

Cuando el número de task varía, puedes usar tokio::task::JoinSet. Te permite obtener resultados en el orden en que los task se completan usando join_set.join_next().await.

5. Ejecución concurrente en el mismo task: join!, select!, timeout

5.1. join! y try_join!

No siempre necesitas spawn. tokio::join! hace poll a múltiples future concurrentemente dentro del mismo task y espera a que todos terminen:

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!("Cliente {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 - Cliente 7 (200 ms)

El tiempo total es el del future más lento, aproximadamente 200 ms. Los future no necesitan ser 'static, así que pueden pedir prestadas variables locales.

Cuando las funciones devuelven Result, usas tokio::try_join!. Esta macro se detiene en el primer error y descarta los future restantes:

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

Aquí fetch_order_checked y fetch_customer_checked son funciones async que devuelven 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!("Se acabó el tiempo de espera"),
}

Se acabó el tiempo de espera

fetch_order necesita 200 ms, superando el límite de 50 ms. En ese caso, timeout devuelve Err(Elapsed) y el future dentro es descartado, es decir, cancelado.

5.3. select!

select! espera a que se complete la rama primera y luego cancela las ramas restantes:

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!("Procesado {order}"),
        _ = shutdown_rx => println!("Recibida señal de shutdown, parando"),
    }
}

Recibida señal de shutdown, parando

La señal de shutdown llega después de 100 ms, antes que fetch_order (200 ms), así que la segunda rama gana.

Nota: Cancelar en Rust significa que el future es descartado en el punto .await más cercano. Si la rama cancelada estaba a mitad de una operación no atómica, por ejemplo ya había leído la mitad de un mensaje del socket, los datos pueden perderse. Deberías revisar la sección “cancel safety” en la documentación de cada función antes de ponerla en select! dentro de un bucle.

6. Compartir datos entre task

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!("Quedan: {}", inventory.lock().unwrap()["laptop"]);
}

Quedan: 5

Aquí usamos std::sync::Mutex, no tokio::sync::Mutex. Esta es la opción recomendada cuando la sección crítica es corta y no hay .await dentro, porque el mutex estándar es más rápido.

Solo necesitas tokio::sync::Mutex cuando tienes que mantener el lock a través de un punto .await, por ejemplo cuando sostienes una conexión compartida mientras esperas I/O.

6.2. Channel mpsc

En lugar de compartir memoria, puedes pasar mensajes entre task usando un channel:

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); // cierra el sender original para que el bucle de abajo termine

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

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

El orden de recepción puede variar entre ejecuciones. El parámetro 32 es la capacidad del buffer. Cuando el buffer está lleno, send espera, creando un mecanismo de back-pressure natural. recv() devuelve None cuando todos los sender han sido descartados. Si olvidas drop(tx), el programa se quedará colgado en el bucle while.

Además de mpsc, Tokio también tiene:

• oneshot: envía un único valor.
• broadcast: múltiples consumidores, cada uno recibe todos los mensajes.
• watch: el consumidor solo le importa el valor más reciente.

7. Manejar código blocking

Tokio usa cooperative scheduling. Los task solo ceden el thread en los puntos .await. Una operación CPU pesada o I/O sincrónico ocupará el worker thread y hará que otros task se queden sin recursos.

fn hash_password(password: &str) -> String {
    // simula una operación que consume CPU, por ejemplo 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 blocking entró en pánico");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking mueve el closure a un thread pool separado destinado a operaciones blocking. Así, los worker thread del runtime siguen disponibles para procesar otras solicitudes.

Nota: No llames a std::thread::sleep, std::fs::read u cliente HTTP sincrónico directamente dentro de una función async. Usa la versión async (tokio::time::sleep, tokio::fs) o envuélvelos en spawn_blocking. Para cálculos en paralelo a gran escala, Rayon suele ser más apropiado.

8. Testear código async

La macro #[tokio::test] crea un runtime para cada test. Por defecto es un runtime de un solo 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 (requiere feature test-util) congela el reloj de Tokio. Cuando todos los task están esperando un timer, el runtime salta automáticamente al siguiente punto en el tiempo. Así, los tests sobre timeout se ejecutan casi instantáneamente y dan resultados estables, sin depender de la velocidad de la máquina CI.

Si necesitas verificar el comportamiento en múltiples threads, usa #[tokio::test(flavor = "multi_thread", worker_threads = 2)].

9. Errores comunes

Mantener un std::sync::MutexGuard a través de .await. Este guard no es Send, así que el future tampoco lo es. Cuando haces tokio::spawn, el compilador se queja con future cannot be sent between threads safely. La solución es reducir el scope del guard con un bloque { ... } antes del .await, o cambiar a tokio::sync::Mutex.

Crear runtime anidados. Llamar a Runtime::new().block_on(...) dentro de código que ya se ejecuta en Tokio causará pánico con el mensaje Cannot start a runtime from within a runtime. En contexto async, solo necesitas .await.

Usar async cuando no es necesario. Una herramienta CLI que lee unos pocos archivos o un job puramente computacional no se benefician de Tokio. Solo pagas el precio de más tiempo de compilación y complejidad. Async brilla cuando hay muchas operaciones I/O concurrentes: servidores web, crawlers, proxies, clientes que llaman a múltiples servicios.

Spawn sin límite. Los task son baratos, pero no gratis. Crear un task por cada elemento en una lista de un millón de ítems puede agotar recursos aguas abajo, por ejemplo el pool de conexiones de la BD. Limita la concurrencia usando tokio::sync::Semaphore o buffer_unordered del crate futures.

Ignorar JoinError. Si haces .await al handle pero usas let _ =, un pánico dentro del task se tragará silenciosamente. Deberías registrarlo o propagarlo hacia arriba.

10. Conclusión

Este artículo ha cubierto los puntos clave del async en Rust con Tokio:

• Los Future son lazy y solo se ejecutan cuando se les hace .await.
• tokio::spawn crea task que se ejecutan concurrentemente. El future pasado debe cumplir Send + 'static.
• join! y try_join! ejecutan múltiples future en el mismo task. select! y timeout manejan carreras y cancelación.
• Los datos compartidos se pueden intercambiar a través de Arc<Mutex<T>> o pasar por channel como mpsc y oneshot.
• El código blocking debe ir en spawn_blocking.
• #[tokio::test(start_paused = true)] hace que los tests sobre tiempo sean rápidos y estables.

Una vez que domines estos conceptos, puedes continuar con I/O de red mediante tokio::net o construir servicios web en Tokio usando frameworks como Axum.