use tokio::sync::Semaphore;

static DB_SEMAPHORE: once_cell::sync::Lazy<Semaphore> = 
    once_cell::sync::Lazy::new(|| Semaphore::new(100));

async fn query_db(sql: &str) -> Result<Vec<Row>> {
    let _permit = DB_SEMAPHORE.acquire().await?;
    // 确保最多只有100个并发数据库查询
    sqlx::query_as::<_, Row>(sql).fetch_all(&pool).await
}

第三章 异步同步原语与通信模式

3.1 同步原语选型指南

Tokio提供了多种同步原语，选型错误可能导致死锁或性能问题：

3.2 Channel通信模式

tokio::sync::mpsc和broadcast是最常用的通信模式：

// 有界channel：背压控制的核心
let (tx, mut rx) = tokio::sync::mpsc::channel::<Event>(1024);

// 多生产者-多消费者
let (tx, rx) = tokio::sync::broadcast::channel::<LogMessage>(256);

// oneshot：请求-响应模式
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel::<Result<Data>>();
// 发送端发送一个值
resp_tx.send(Ok(data)).ok();
// 接收端等待单个值
let result = resp_rx.await.unwrap();

3.3 异步死锁的排查与预防

生产环境中最常见的异步死锁模式：

// 错误模式1：在持有锁的情况下await
async fn deadlock_bad(mutex: &tokio::sync::Mutex<State>) {
    let mut guard = mutex.lock().await;
    let data = some_async_io().await;  // 锁跨越.await！其他任务无法获取锁
    guard.update(data);
}

// 正确做法：缩小锁持有范围
async fn deadlock_good(mutex: &tokio::sync::Mutex<State>) {
    let data = some_async_io().await;  // 先完成异步操作
    let mut guard = mutex.lock().await;  // 再获取锁，立即更新释放
    guard.update(data);
}

排查异步死锁的工具链：

第四章 异步流（Stream）与背压处理

4.1 Stream trait与Iterator的对比

Stream是异步版本的Iterator，核心差异在于next()返回Poll>而非Option：

use tokio_stream::{Stream, StreamExt};

// 自定义Stream：模拟一个异步数据源
struct SensorStream {
    receiver: tokio::sync::mpsc::Receiver<SensorData>,
}

impl Stream for SensorStream {
    type Item = SensorData;
    
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.receiver.poll_recv(cx)
    }
}

// 使用Stream进行背压控制的数据处理管道
async fn process_sensor_pipeline(mut stream: SensorStream) {
    stream
        .filter(|d| futures::future::ready(d.temperature > 20.0))
        .map(|d| d.normalize())
        .chunks_timeout(100, std::time::Duration::from_millis(50))
        .for_each(|batch| async {
            // 每批100条或50ms超时批量写入数据库
            db.insert_batch(&batch).await;
        })
        .await;
}

4.2 背压在生产中的实践

在高吞吐系统中，生产者速度可能远超消费者。忽略背压会导致OOM或服务不可用：

// 有界channel实现生产级背压
let (tx, rx) = tokio::sync::mpsc::channel::<Task>(8192);

// 生产者：当channel满时等待，不会无限制累积
async fn producer(tx: tokio::sync::mpsc::Sender<Task>) {
    for task in incoming_requests {
        // send().await 会在channel满时阻塞，天然实现背压
        if tx.send(task).await.is_err() {
            error!("Channel closed, stopping producer");
            break;
        }
    }
}

// 消费者池：多个消费者并行处理
async fn consumer_pool(rx: tokio::sync::mpsc::Receiver<Task>, pool_size: usize) {
    let pool: Vec<_> = (0..pool_size)
        .map(|id| {
            let mut rx = rx.resubscribe(); // 假设broadcast
            tokio::spawn(consumer(id, rx))
        })
        .collect();
    
    for handle in pool {
        handle.await.unwrap();
    }
}

第五章 生产级网络服务的异步架构

5.1 Hyper HTTP服务器的异步实践

use axum::{
    routing::{get, post},
    Router,
    extract::{State, Json},
    response::IntoResponse,
};
use std::sync::Arc;
use tokio::sync::RwLock;

// 应用状态
struct AppState {
    db_pool: sqlx::PgPool,
    cache: Arc<RwLock<lru::LruCache<String, Vec<u8>>>>,
    rate_limiter: Arc<RateLimiter>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // tracing初始化
    tracing_subscriber::fmt()
        .with_env_filter("info,tower_http=debug")
        .with_target(false)
        .init();

    let state = Arc::new(AppState {
        db_pool: sqlx::postgres::PgPoolOptions::new()
            .max_connections(100)
            .connect(&std::env::var("DATABASE_URL")?)
            .await?,
        cache: Arc::new(RwLock::new(lru::LruCache::new(10_000))),
        rate_limiter: Arc::new(RateLimiter::new(1000, std::time::Duration::from_secs(60))),
    });

    let app = Router::new()
        .route("/api/data/:id", get(get_data))
        .route("/api/data", post(create_data))
        .layer(tower::ServiceBuilder::new()
            .layer(tower_http::trace::TraceLayer::new_for_http())
            .layer(tower_http::compression::CompressionLayer::new())
            .layer(tower_http::limit::RequestBodyLimitLayer::new(10 * 1024 * 1024))
            .timeout(std::time::Duration::from_secs(30))
            .into_inner())
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    info!("Server listening on :3000");
    axum::serve(listener, app).await?;
    Ok(())
}

async fn get_data(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<impl IntoResponse, AppError> {
    // 先查缓存
    {
        let cache = state.cache.read().await;
        if let Some(data) = cache.get(&id) {
            return Ok(([(axum::http::header::CONTENT_TYPE, "application/json")], data.clone()));
        }
    }

    // 缓存未命中，查询数据库
    let row = sqlx::query_as::<_, DataRow>("SELECT * FROM data WHERE id = $1")
        .bind(&id)
        .fetch_optional(&state.db_pool)
        .await?
        .ok_or(AppError::NotFound)?;

    // 更新缓存
    {
        let mut cache = state.cache.write().await;
        let serialized = serde_json::to_vec(&row)?;
        cache.put(id, serialized);
    }

    Ok(Json(row))
}

5.2 gRPC异步服务构建

use tonic::{Request, Response, Status};

pub mod pb {
    tonic::include_proto!("helloworld");
}

use pb::greeter_server::{Greeter, GreeterServer};
use pb::{HelloReply, HelloRequest};

pub struct MyGreeter {
    db_pool: sqlx::PgPool,
}

#[tonic::async_trait]
impl Greeter for MyGreeter {
    async fn say_hello(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<HelloReply>, Status> {
        let name = request.into_inner().name;

        // 异步数据库查询
        let greeting = sqlx::query_scalar::<_, String>(
            "SELECT greeting FROM greetings WHERE name = $1"
        )
        .bind(&name)
        .fetch_optional(&self.db_pool)
        .await
        .map_err(|e| Status::internal(format!("DB error: {}", e)))?
        .unwrap_or_else(|| format!("Hello, {}!", name));

        Ok(Response::new(HelloReply { message: greeting }))
    }
}

// gRPC流处理：实时数据推送
#[tonic::async_trait]
impl DataService for MyDataService {
    type SubscribeStream = tokio_stream::wrappers::ReceiverStream<Result<DataEvent, Status>>;

    async fn subscribe(
        &self,
        request: Request<SubscribeRequest>,
    ) -> Result<Response<Self::SubscribeStream>, Status> {
        let (tx, rx) = tokio::sync::mpsc::channel(128);
        let filter = request.into_inner();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_millis(100));
            loop {
                interval.tick().await;
                let event = generate_data_event(&filter);
                if tx.send(Ok(event)).await.is_err() {
                    break; // 客户端断开
                }
            }
        });

        Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx)))
    }
}

5.3 优雅关闭（Graceful Shutdown）

生产环境的优雅关闭必须确保：

use tokio::signal;
use std::time::Duration;

async fn run_with_graceful_shutdown(app: Router, port: u16) -> Result<(), Box<dyn Error>> {
    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
    
    let server = axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal());

    info!("Server started on port {}", port);
    
    // 等待服务器完成或出错
    if let Err(e) = server.await {
        error!("Server error: {}", e);
    }

    // 释放应用资源
    info!("Server shutdown complete");
    Ok(())
}

async fn shutdown_signal() {
    let ctrl_c = async {
        signal::ctrl_c().await.expect("Failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("Failed to install SIGTERM handler")
            .recv().await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => { info!("Received Ctrl+C, shutting down...") },
        _ = terminate => { info!("Received SIGTERM, shutting down...") },
    }

    // 给优雅关闭一个超时期限
    tokio::spawn(async {
        tokio::time::sleep(Duration::from_secs(30)).await;
        error!("Graceful shutdown timed out after 30s, forcing exit");
        std::process::exit(1);
    });
}

第六章 性能优化与调试技巧

6.1 异步性能基准与对比

在AWS c6g.2xlarge（ARM Neoverse-N1, 8核）上的HTTP服务基准测试：

Rust异步HTTP服务在吞吐量和延迟上全面领先，且内存占用最低。

6.2 异步代码性能陷阱

陷阱一：不必要的内存分配

// 每次调用都分配新的String
async fn handle_bad(req: Request) -> Response {
    let body = req.body().await;           // 分配
    let processed = body.to_uppercase();    // 又分配
    Response::new(processed)
}

// 复用内存：使用&str或Cow
async fn handle_good(req: Request) -> Response {
    let body = req.body().await;
    // 只在必要时分配
    Response::new(processed.into_owned())
}

// 不必要地每个handler都克隆Arc链
async fn handler_bad(
    State(db): State<Arc<PgPool>>,
    State(cache): State<Arc<RwLock<Cache>>>,
    State(limiter): State<Arc<RateLimiter>>,
) -> Response {
    // 每次3次Arc::clone（原子操作）
}

// 更好的做法：合并状态为一个Arc
struct AppState { db: PgPool, cache: RwLock<Cache>, limiter: RateLimiter }
// 只需一次clone

// 全局Mutex保护的计数器成为瓶颈
static COUNTER: Lazy<Mutex<HashMap<String, u64>>> = Lazy::new(Default::default);

// 替换为分片锁减少竞争
use dashmap::DashMap;
static COUNTER_SHARDED: Lazy<DashMap<String, u64>> = Lazy::new(Default::default);
// DashMap内部使用RwLock数组分片，大幅降低锁竞争

// 在Cargo.toml添加依赖
// tokio = { version = "1", features = ["full", "tracing"] }
// console-subscriber = "0.2"

#[tokio::main]
async fn main() {
    // 开启console subscriber
    console_subscriber::init();
    
    // ... 应用代码
}

// 另终端运行: tokio-console
// 可实时查看:
// - 所有活跃任务及其状态（idle/polling/blocked）
// - 每个任务的poll耗时分布
// - I/O资源使用情况
// - 同步原语（Mutex/Semaphore）等待队列

use async_trait::async_trait;

#[async_trait]
trait DataRepository: Send + Sync {
    async fn find_by_id(&self, id: &str) -> Result<Option<Data>, Error>;
    async fn save(&self, data: &Data) -> Result<(), Error>;
    async fn delete(&self, id: &str) -> Result<bool, Error>;
}

// 实现可以是PostgreSQL、MySQL、内存等
#[async_trait]
impl DataRepository for PostgresRepo {
    async fn find_by_id(&self, id: &str) -> Result<Option<Data>, Error> {
        sqlx::query_as::<_, Data>("SELECT * FROM data WHERE id = $1")
            .bind(id)
            .fetch_optional(&self.pool)
            .await
            .map_err(Into::into)
    }
    
    async fn save(&self, data: &Data) -> Result<(), Error> {
        sqlx::query("INSERT INTO data (id, value) VALUES ($1, $2)")
            .bind(&data.id)
            .bind(&data.value)
            .execute(&self.pool)
            .await?;
        Ok(())
    }
    
    async fn delete(&self, id: &str) -> Result<bool, Error> {
        let result = sqlx::query("DELETE FROM data WHERE id = $1")
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }
}

// 使用impl Future避免Box::pin（Rust 1.75+）
// 注：目前仍需要async_trait宏，但新版已开始支持

// 或手动实现trait（无宏开销）
trait FastAsync {
    fn poll_find(self: Pin<&mut Self>, cx: &mut Context<'_>, id: &str) 
        -> Poll<Result<Option<Data>, Error>>;
}

HTTP客户端:   reqwest (tokio原生) > hyper (底层控制)
序列化:       serde + serde_json (标准)
数据库:       sqlx (编译时检查) > diesel-async (ORM风格)
消息队列:     rdkafka (Kafka) / lapin (RabbitMQ)
缓存:         redis (官方) / moka (本地缓存)
日志:         tracing (异步感知) > log
配置:         config / figment
测试:         tokio::test + mockall