use tokio_stream::{self as Stream, StreamExt};

async fn process_stream(stream: impl Stream<Item = Data>) {
    stream
        .chunks(100)           // 批量处理
        .for_each(|batch| async {
            process_batch(batch).await;
        })
        .await;
}

use tokio::sync::Semaphore;

let semaphore = Semaphore::new(10);  // 最多 10 个并发

let stream = tokio_stream::iter(0..1000)
    .map(|i| {
        let permit = semaphore.clone().acquire_owned();
        async move {
            let _permit = permit.await;
            process(i).await
        }
    })
    .buffer_unordered(100);  // 最多 100 并发

use tokio::select;
use tokio::time::{sleep, timeout};

async fn multiplex() {
    loop {
        select! {
            msg = receiver.recv() => {
                if let Ok(msg) = msg {
                    handle(msg).await;
                }
            }
            _ = sleep(Duration::from_secs(5)) => {
                // 超时处理
            }
            else => break,  // 所有分支都完成
        }
    }
}

use tokio::time::timeout;

async fn with_timeout() -> Result<Value, TimeoutError> {
    timeout(Duration::from_secs(5), long_operation()).await
}

// 协作式取消
let mut task = tokio::spawn(async move {
    loop {
        // 检查取消
        if task.is_cancelled() {
            return;
        }
        // 继续工作
    }
});

// 取消任务
task.abort();

// 并行执行，不等待完成
let (a, b) = tokio::join!(async_a(), async_b());

// 全部成功才成功
let (a, b) = tokio::try_join!(async_a(), async_b())?;

// 错误传播
fn combined() -> impl Future<Output = Result<(A, B), E>> {
    async {
        let (a, b) = try_join!(op_a(), op_b())?;
        Ok((a, b))
    }
}