使用 Rocket 和 Python SciPy 构建处理密集计算的 ISR 架构实践

一个棘手的问题摆在面前：我们需要为一组复杂的物理模拟数据提供一个Web可视化界面。这些模拟结果的计算非常耗时，Python端的SciPy和NumPy脚本运行一次需要10到30秒才能生成一份完整的数据集。如果采用传统的服务器端渲染（SSR），每个用户请求都意味着长达数十秒的白屏等待，这在生产环境中是完全不可接受的。

最初的方案是全量预生成所有可能的模拟结果并缓存为静态页面。但这很快被否决，因为模拟的输入参数组合是近乎无限的，存储成本和管理复杂度会失控。增量静态再生（Incremental Static Regeneration, ISR）的思路似乎是破局之道：服务一个缓存的旧版本页面，同时在后台触发一个异步任务来重新生成新版本。这样用户可以立即看到内容，并在下次访问时获得更新后的数据。

然而，我们团队的技术栈是基于Rust的，追求极致的性能和资源控制，并不想为此引入庞大的Node.js生态（如Next.js）。我们的计算服务则坚定地使用Python，因为它拥有无与伦比的科学计算库生态。问题由此转化为：如何在一个异构技术栈（Rust + Python）中，从零开始构建一个健壮、高效的ISR服务。

我们的目标架构如下：

1. Web服务层: 使用Rust的Rocket框架，它负责接收用户请求，并执行ISR核心逻辑。
2. 计算层: 一个独立的Python微服务，使用FastAPI封装，暴露一个接口，接收计算参数并调用SciPy执行密集型任务。
3. 通信: Rocket服务通过HTTP请求调用Python计算服务。
4. 存储: 生成的HTML页面直接存储在文件系统上，作为静态资源。

graph TD
    subgraph "用户浏览器"
        A[Client Request]
    end

    subgraph "Rust 服务 (Rocket)"
        B(Rocket Route Handler)
        C{Cache State Check}
        D[Serve Static HTML]
        E(Spawn Tokio Task)
        F[Render Template & Write File]
        G(HTTP Client: reqwest)
        H[In-Memory State Manager]
    end

    subgraph "Python 服务 (FastAPI)"
        I(FastAPI Endpoint)
        J[SciPy/NumPy Computation]
    end

    A --> B
    B --> C
    C -- Fresh/Stale --> D
    D --> A
    C -- Stale/Generating --> E
    C -- Not Found --> E
    B --> H
    E --> H
    E --> G
    G --> I
    I --> J
    J --> I
    I --> G
    G --> F
    F --> D
    F --> H
    H --> C

这个架构的核心挑战在于Rust端的ISR逻辑实现。它必须是线程安全的，能够处理并发请求，避免对同一资源发起重复的后台计算（惊群效应），并优雅地处理计算失败或超时的场景。

第一步：构建Python计算服务

这部分相对直接。我们使用FastAPI创建一个简单的Web服务。为了模拟真实的计算耗时，我们加入asyncio.sleep。在真实项目中，这里会是复杂的SciPy和NumPy矩阵运算。

文件 computation_service/main.py:

import asyncio
import time
from typing import List

import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

# 配置
# 在生产中, 这应该来自环境变量
SIMULATION_BASE_DELAY_SECONDS = 5.0
SIMULATION_COMPLEXITY_FACTOR = 0.01

app = FastAPI(
    title="Scientific Computation Service",
    description="A service to perform intensive SciPy/NumPy simulations.",
    version="1.0.0",
)

# 定义请求和响应模型，这是良好API设计的关键
class SimulationParams(BaseModel):
    grid_size: int = Field(
        100, gt=10, le=500, description="The size of the square grid for simulation."
    )
    frequency: float = Field(
        5.0, gt=0, description="Frequency parameter for the wave function."
    )
    damping: float = Field(
        0.1, gt=0, le=1.0, description="Damping factor for the simulation."
    )

class SimulationResult(BaseModel):
    computation_time_ms: float
    grid_size: int
    data: List[List[float]]
    description: str

@app.post("/compute", response_model=SimulationResult)
async def run_computation(params: SimulationParams):
    """
    接收模拟参数，执行计算并返回结果。
    这是一个计算密集型任务，我们用numpy和sleep来模拟。
    """
    start_time = time.perf_counter()
    
    # 这里的日志在生产环境中至关重要
    print(f"Received computation request with params: {params.json()}")

    try:
        # 模拟计算延迟，延迟时间与网格大小相关
        # 这是为了防止非常大的请求阻塞服务太久
        dynamic_delay = SIMULATION_BASE_DELAY_SECONDS + \
                        (params.grid_size ** 2) * SIMULATION_COMPLEXITY_FACTOR / 1000
        await asyncio.sleep(dynamic_delay)

        # 核心计算：使用 NumPy 生成一个波形表面
        # 这部分可以替换为任何复杂的 SciPy 运算
        x = np.linspace(-10, 10, params.grid_size)
        y = np.linspace(-10, 10, params.grid_size)
        xx, yy = np.meshgrid(x, y)
        
        # 一个复杂的数学函数
        r = np.sqrt(xx**2 + yy**2) + 1e-9 # 避免除以零
        zz = np.sin(params.frequency * r) / r * np.exp(-params.damping * r)
        
        # 将numpy数组转换为Python列表以便JSON序列化
        data_list = zz.tolist()

        end_time = time.perf_counter()
        computation_time = (end_time - start_time) * 1000

        print(f"Computation finished in {computation_time:.2f} ms for grid_size={params.grid_size}")

        return SimulationResult(
            computation_time_ms=computation_time,
            grid_size=params.grid_size,
            data=data_list,
            description=f"Damped wave simulation with frequency {params.frequency}",
        )
    except Exception as e:
        # 健壮的错误处理
        print(f"ERROR: Computation failed. Reason: {str(e)}")
        raise HTTPException(
            status_code=500, 
            detail=f"Internal server error during computation: {str(e)}"
        )

if __name__ == "__main__":
    import uvicorn
    # 在生产环境中，你会使用Gunicorn等多进程管理器
    uvicorn.run(app, host="0.0.0.0", port=8001)

这个Python服务已经具备了基本的生产级特性：通过Pydantic进行输入验证、详细的日志输出以及稳健的错误处理。

第二步：Rocket ISR服务的核心设计与实现

这是整个项目的核心。我们需要在Rocket中实现一个状态管理器，以协调对静态页面的生成和访问。

首先，定义页面缓存的状态。一个页面可以处于以下几种状态：

• Fresh: 页面存在且未过期。
• Stale: 页面存在但已过期，可以服务给用户，但需要触发后台更新。
• Generating: 页面正在后台生成中，新来的请求不应触发新的生成任务。
• NotFound: 页面从未生成过。

我们需要一个全局、线程安全的数据结构来存储所有被管理页面的状态。Arc<Mutex<HashMap<...>>> 是Rust中实现这一目标的标准模式。

isr_service/src/main.rs:

```rust
#[macro_use]
extern crate rocket;

use rocket::fs::NamedFile;
use rocket::http::Status;
use rocket::serde::json::Json;
use rocket::serde::{Deserialize, Serialize};
use rocket::State;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::fs;
use tokio::sync::Mutex;
use tera::{Context, Tera};

// — 配置 —
const STALE_AFTER_SECONDS: u64 = 60; // 页面被视为“陈旧”的时间
const STATIC_DIR: &str = “static_pages”; // 存储生成HTML的目录
const PYTHON_SERVICE_URL: &str = “http://127.0.0.1:8001/compute";

// — 状态管理 —

#[derive(Clone, Debug, PartialEq)]
enum PageStatus {
Fresh,
Stale,
Generating,
}

#[derive(Clone, Debug)]
struct PageState {
status: PageStatus,
last_generated: SystemTime,
}

// 全局状态管理器
type CacheManager = Arc<Mutex<HashMap<String, Arc<Mutex>>>>;

// — 数据模型 —

#[derive(Serialize)]
struct SimulationParams {
grid_size: u32,
frequency: f32,
damping: f32,
}

#[derive(Deserialize, Debug)]
struct SimulationResult {
computation_time_ms: f64,
grid_size: u32,
data: Vec<Vec>,
description: String,
}

// — Rocket 路由和核心逻辑 —

#[get(“/view///“)]
async fn get_visualization(
grid_size: u32,
frequency: f32,
damping: f32,
manager: &State,
tera: &State,
) -> Result<NamedFile, Status> {
// 1. 确定资源标识符和文件路径
let resource_id = format!(“{}{}{}”, grid_size, frequency, damping);
let file_path = PathBuf::from(STATIC_DIR).join(format!(“{}.html”, resource_id));

// 2. 检查文件是否存在
if !file_path.exists() {
    // 文件不存在，必须立即生成
    info!("'{}' not found. Triggering initial generation.", resource_id);
    trigger_generation(resource_id.clone(), grid_size, frequency, damping, manager.inner().clone(), tera.inner().clone(), true).await;
}

// 3. 获取或创建页面状态锁
let page_state_arc = {
    let mut manager_guard = manager.lock().await;
    manager_guard
        .entry(resource_id.clone())
        .or_insert_with(|| {
            Arc::new(Mutex::new(PageState {
                status: PageStatus::Fresh, // 假设存在的文件初始是 Fresh
                last_generated: SystemTime::now(),
            }))
        })
        .clone()
};

// 4. 分析当前状态并决策
{
    let mut state = page_state_arc.lock().await;
    let elapsed = state.last_generated.elapsed().unwrap_or_default();

    if state.status != PageStatus::Generating && elapsed > Duration::from_secs(STALE_AFTER_SECONDS) {
        state.status = PageStatus::Stale;
    }

    match state.status {
        PageStatus::Fresh => {
            info!("'{}' is Fresh. Serving static file.", resource_id);
        }
        PageStatus::Stale => {
            info!("'{}' is Stale. Serving static file and triggering background regeneration.", resource_id);
            // 状态改为 Generating 防止其他请求重复触发
            state.status = PageStatus::Generating;
            // 关键：在后台线程中执行，不阻塞当前请求
            let manager_clone = manager.inner().clone();
            let tera_clone = tera.inner().clone();
            tokio::spawn(async move {
                trigger_generation(resource_id, grid_size, frequency, damping, manager_clone, tera_clone, false).await;
            });
        }
        PageStatus::Generating => {
            info!("'{}' is already generating. Serving stale file.", resource_id);
        }
    }
}

// 5. 无论状态如何（除了首次生成），都尝试服务文件
NamedFile::open(file_path).await.map_err(|e| {
    error!("Failed to open file for '{}': {}", resource_id, e);
    Status::InternalServerError
})

}

// — 后台生成任务 —

async fn trigger_generation(
resource_id: String,
grid_size: u32,
frequency: f32,
damping: f32,
manager: CacheManager,
tera: Tera,
is_initial: bool, // 标记是否是阻塞的首次生成
) {
info!(“Starting generation for ‘{}’…”, resource_id);

// a. 准备请求参数
let params = SimulationParams { grid_size, frequency, damping };

// b. 调用Python服务
let client = reqwest::Client::new();
let res = match client.post(PYTHON_SERVICE_URL).json(&params).send().await {
    Ok(res) => res,
    Err(e) => {
        error!("Failed to connect to Python service for '{}': {}", resource_id, e);
        update_state_on_failure(resource_id, manager).await;
        return;
    }
};

// c. 处理Python服务响应
if res.status().is_success() {
    match res.json::<SimulationResult>().await {
        Ok(result) => {
            // d. 渲染HTML模板
            let mut context = Context::new();
            context.insert("params", &params);
            context.insert("result", &result);
            let rendered_html = match tera.render("visualization.html.tera", &context) {
                Ok(html) => html,
                Err(e) => {
                    error!("Failed to render template for '{}': {}", resource_id, e);
                    update_state_on_failure(resource_id, manager).await;
                    return;
                }
            };

            // e. 原子化写入文件
            let file_path = Path::new(STATIC_DIR).join(format!("{}.html", resource_id));
            let temp_path = file_path.with_extension("html.tmp");
            
            if let Err(e) = fs::write(&temp_path, rendered_html).await {
                error!("Failed to write to temp file for '{}': {}", resource_id, e);
                update_state_on_failure(resource_id, manager).await;
                return;
            }
            if let Err(e) = fs::rename(&temp_path, &file_path).await {
                error!("Failed to rename temp file for '{}': {}", resource_id, e);
                update_state_on_failure(resource_id, manager).await;
                return;
            }

            info!("Successfully generated and saved '{}'.", resource_id);

            // f. 更新状态为Fresh
            let page_state_arc = manager.lock().await.get(&resource_id).unwrap().clone();
            let mut state = page_state_arc.lock().await;
            state.status = PageStatus::Fresh;
            state.last_generated = SystemTime::now();
        }
        Err(e) => {
            error!("Failed to parse JSON from Python service for '{}': {}", resource_id, e);
            update_state_on_failure(resource_id, manager).await;
        }
    }
} else {
    let status = res.status();
    let body = res.text().await.unwrap_or_else(|_| "Could not read body".to_string());
    error!(
        "Python service returned error for '{}': Status {}, Body: {}",
        resource_id, status, body
    );
    update_state_on_failure(resource_id, manager).await;
}

}

// 辅助函数：处理生成失败，重置状态
async fn update_state_on_failure(resource_id: String, manager: CacheManager) {
let page_state_arc_opt = manager.lock().await.get(&resource_id).cloned();
if let Some(page_state_arc) = page_state_arc_opt {
let mut state = page_state_arc.lock().await;
// 如果失败，我们将其状态重置为Stale，以便下次请求可以重新触发
// 但为了防止快速连续失败导致的服务过载，可以加入冷却逻辑
state.status = PageStatus::Stale;
warn!(“Reset status to Stale for ‘{}’ after generation failure.”, resource_id);
}
}

#[launch]
async fn rocket() -> _ {
// 确保静态目录存在
fs::create_dir_all(STATIC_DIR).await.expect(“Failed to create static directory”);

// 初始化Tera模板引擎
let tera = Tera::new("templates/**/*").expect("Failed to parse templates");

// 初始化状态管理器
let cache_manager = Arc::new(Mutex::new(HashMap::new()));

rocket::build()
    .manage(cache_manager)
    .manage(tera)
    .mount("/", routes