From f6c9141449e9785d290407ecc2d0b752be16824b Mon Sep 17 00:00:00 2001 From: lychang <632945494@qq.com> Date: Wed, 25 Mar 2026 18:31:11 +0800 Subject: [PATCH] update code level --- Cargo.lock | 1 + Cargo.toml | 3 +- config.toml | 6 + src/config.rs | 9 + src/image_processor.rs | 36 +- src/import.rs | 83 +++-- src/main.rs | 39 ++- src/models.rs | 58 ++-- src/queue.rs | 723 +++++++++++++++++++++++++++++++++++++++++ src/routes.rs | 106 +++--- 10 files changed, 962 insertions(+), 102 deletions(-) create mode 100644 src/queue.rs diff --git a/Cargo.lock b/Cargo.lock index 7aff252..715c5c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1236,6 +1236,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "thiserror", "tokio", "tracing", "tracing-subscriber", diff --git a/Cargo.toml b/Cargo.toml index bc3ae18..192ac88 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,4 +17,5 @@ reqwest = { version = "0.11", default-features = false, features = ["blocking"," config = "0.13" rand = "0.8" tracing = "0.1" -tracing-subscriber = "0.3" \ No newline at end of file +tracing-subscriber = "0.3" +thiserror = "1.0" \ No newline at end of file diff --git a/config.toml b/config.toml index b6fe01e..2ca5044 100644 --- a/config.toml +++ b/config.toml @@ -8,3 +8,9 @@ port = 8080 [storage] images_dir = "./albums" thumbnails_dir = "./thumbnails" + +[queue] +max_concurrent_tasks = 6 +queue_size_limit = 200 +high_request_rate_threshold = 200.0 +idle_request_rate_threshold = 10.0 \ No newline at end of file diff --git a/src/config.rs b/src/config.rs index fd7f43e..bde341f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,6 +7,7 @@ pub struct Config { pub database: DatabaseConfig, pub server: ServerConfig, pub storage: StorageConfig, + pub queue: QueueConfig, } @@ -27,6 +28,14 @@ pub struct StorageConfig { pub thumbnails_dir: PathBuf, } +#[derive(Debug, Deserialize, Clone)] +pub struct QueueConfig { + pub max_concurrent_tasks: usize, + pub queue_size_limit: usize, + pub high_request_rate_threshold: f64, + pub idle_request_rate_threshold: f64, +} + impl Config { pub fn load() -> Result { let config = config::Config::builder() diff --git a/src/image_processor.rs b/src/image_processor.rs index f497908..f528a38 100644 --- a/src/image_processor.rs +++ b/src/image_processor.rs @@ -3,6 +3,30 @@ use std::fs; use image::{GenericImageView, ImageFormat}; use uuid::Uuid; use crate::config::StorageConfig; +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum ImageProcessorError { + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + #[error("Image processing error: {0}")] + Image(#[from] image::ImageError), + #[error("Invalid image data")] + InvalidData, +} + +// 手动实现 From 以支持 ? 操作符 +impl From for ImageProcessorError { + fn from(err: reqwest::Error) -> Self { + ImageProcessorError::Io(std::io::Error::new(std::io::ErrorKind::Other, err.to_string())) + } +} + +impl ImageProcessorError { + pub fn is_send(&self) -> bool { + true + } +} pub struct ImageProcessor { pub config: StorageConfig, @@ -17,7 +41,7 @@ impl ImageProcessor { Self { config } } - pub async fn save_image(&self, image_data: &[u8], filename: &str) -> Result> { + pub async fn save_image(&self, image_data: &[u8], filename: &str) -> Result { let image_id = Uuid::new_v4(); let file_extension = Path::new(filename) .extension() @@ -56,7 +80,7 @@ impl ImageProcessor { output_path: &Path, _max_width: u32, _max_height: u32, - ) -> Result<(), Box> { + ) -> Result<(), ImageProcessorError> { // 使用统一的缩略图尺寸（400x300，保持宽高比） let thumbnail = img.thumbnail(400, 300); @@ -81,7 +105,7 @@ impl ImageProcessor { } /// 快速生成缩略图（低质量但快速） - pub async fn generate_fast_thumbnail(&self, image_data: &[u8], _filename: &str) -> Result, Box> { + pub async fn generate_fast_thumbnail(&self, image_data: &[u8], _filename: &str) -> Result, ImageProcessorError> { // 使用快速解码和缩放 let img = image::load_from_memory(image_data)?; @@ -130,7 +154,7 @@ impl ImageProcessor { &self, url: &str, filename: &str, - ) -> Result> { + ) -> Result { let response = reqwest::get(url).await?; let image_data = response.bytes().await?; @@ -142,7 +166,7 @@ impl ImageProcessor { &self, image_data: &[u8], thumbnail_filename: &str, - ) -> Result<(), Box> { + ) -> Result<(), ImageProcessorError> { // 使用统一的缩略图生成方法 let thumbnail_data = self.generate_thumbnail_from_memory(image_data, thumbnail_filename).await?; @@ -161,7 +185,7 @@ impl ImageProcessor { &self, image_data: &[u8], _thumbnail_filename: &str, - ) -> Result, Box> { + ) -> Result, ImageProcessorError> { // 加载图片 let img = image::load_from_memory(image_data)?; diff --git a/src/import.rs b/src/import.rs index e17687a..2151146 100644 --- a/src/import.rs +++ b/src/import.rs @@ -54,13 +54,8 @@ async fn batch_insert_images_and_set_cover( } } -#[derive(Debug, Deserialize)] -struct AlbumIndex { - title: String, - images: Vec, -} -pub async fn import_existing_data(config: &Config, db: Arc) -> Result<(), Box> { +pub async fn import_existing_data(config: &Config, db: Arc, task_queue: Option>) -> Result<(), Box> { println!("开始扫描图片目录并导入新相册..."); let albums_dir = Path::new(&config.storage.images_dir); @@ -147,32 +142,62 @@ pub async fn import_existing_data(config: &Config, db: Arc) -> Result< return Ok(()); } - // 第二阶段：串行处理相册导入，避免数据库死锁 - // 对于SQLite，高并发写入容易导致死锁，改为串行处理 - let mut total_imported = 0; - let mut successful_albums = 0; - - for (folder_name, path) in album_folders { - println!("📁 开始导入相册: {}", folder_name); + // 如果有队列系统，使用队列异步处理相册导入 + if let Some(queue) = task_queue { + println!("🚀 使用队列系统异步导入相册..."); - match import_single_album(&db, &folder_name, &path).await { - Ok(count) => { - if count > 0 { - successful_albums += 1; - total_imported += count; - println!("✅ 相册 {} 导入完成，共导入 {} 张图片", folder_name, count); - } else { - println!("⚠️ 相册 {} 没有图片需要导入", folder_name); - } - } - Err(e) => { - eprintln!("❌ 相册 {} 导入失败: {}", folder_name, e); - // 继续处理下一个相册，不中断整个导入过程 + for (folder_name, path) in album_folders { + println!("📁 添加相册导入任务: {}", folder_name); + + // 创建相册导入任务 + let relative_path = path.strip_prefix(&config.storage.images_dir) + .unwrap_or(&path) + .to_string_lossy() + .to_string(); + let task = crate::queue::TaskFactory::create_album_import_task( + relative_path, + None, + ); + + // 添加任务到队列 + if let Err(e) = queue.add_task(task).await { + eprintln!("❌ 添加相册导入任务失败: {}", e); + } else { + println!("✅ 相册导入任务已添加到队列: {}", folder_name); } } + + println!("🎉 所有相册导入任务已添加到队列，将由AlbumImportWorker异步处理"); + } else { + // 如果没有队列系统，使用原来的串行处理方式 + println!("⚠️ 队列系统不可用，使用串行处理方式导入相册..."); + + let mut total_imported = 0; + let mut successful_albums = 0; + + for (folder_name, path) in album_folders { + println!("📁 开始导入相册: {}", folder_name); + + match import_single_album(&db, &folder_name, &path).await { + Ok(count) => { + if count > 0 { + successful_albums += 1; + total_imported += count; + println!("✅ 相册 {} 导入完成，共导入 {} 张图片", folder_name, count); + } else { + println!("⚠️ 相册 {} 没有图片需要导入", folder_name); + } + } + Err(e) => { + eprintln!("❌ 相册 {} 导入失败: {}", folder_name, e); + // 继续处理下一个相册，不中断整个导入过程 + } + } + } + + println!("🎉 导入完成！成功导入 {} 个相册，共 {} 张图片", successful_albums, total_imported); } - println!("🎉 导入完成！成功导入 {} 个相册，共 {} 张图片", successful_albums, total_imported); Ok(()) } @@ -308,7 +333,7 @@ async fn import_single_album(db: &Database, folder_name: &str, path: &Path) -> R } /// 快速扫描图片文件 -fn scan_image_files(path: &Path, _folder_name: &str) -> Result, Box> { +pub(crate) fn scan_image_files(path: &Path, _folder_name: &str) -> Result, Box> { let mut image_files = Vec::new(); for entry in fs::read_dir(path)? { @@ -483,7 +508,7 @@ async fn generate_thumbnail_for_image( Ok(()) } Err(e) => { - Err(format!("缩略图生成失败: {}", e).into()) + Err(format!("缩略图生成失败: {:?}", e).into()) } } } \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index eb51eb2..9659042 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,13 +4,14 @@ mod image_processor; mod import; mod models; mod routes; +mod queue; use actix_files as afs; use actix_web::{App, HttpResponse, HttpServer, web, middleware}; use std::fs; use std::sync::Arc; use tokio::time::{interval, Duration}; -use tracing::{info, error, warn, debug}; +use tracing::{info, error}; use crate::models::AlbumsCache; #[actix_web::main] @@ -42,19 +43,46 @@ async fn main() -> std::io::Result<()> { } }; + let image_processor = Arc::new(image_processor::ImageProcessor::new(config.storage.clone())); + + // 初始化智能任务队列 + let task_queue = Arc::new(queue::TaskQueue::new( + config.queue.max_concurrent_tasks, // 最大并发任务数 + config.queue.queue_size_limit, // 队列大小限制 + config.queue.high_request_rate_threshold, // 高请求频率阈值 + config.queue.idle_request_rate_threshold, // 空闲请求频率阈值 + )); + // 在后台线程中自动导入新相册（不阻塞HTTP服务启动） let config_clone = config.clone(); let db_clone = Arc::clone(&db); + let task_queue_clone = Arc::clone(&task_queue); tokio::spawn(async move { info!("[后台任务] 开始扫描图片目录并导入新相册..."); - match import::import_existing_data(&config_clone, db_clone).await { + match import::import_existing_data(&config_clone, db_clone, Some(task_queue_clone)).await { Ok(_) => info!("[后台任务] 相册导入完成"), Err(e) => error!("[后台任务] 相册导入失败: {}", e), } }); - // 初始化图片处理器 - let image_processor = Arc::new(image_processor::ImageProcessor::new(config.storage.clone())); + // 启动缩略图生成Worker + let thumbnail_worker = queue::ThumbnailWorker::new( + Arc::clone(&task_queue), + Arc::clone(&image_processor), + ); + tokio::spawn(async move { + thumbnail_worker.start().await; + }); + + // 启动相册导入Worker + let album_import_worker = queue::AlbumImportWorker::new( + Arc::clone(&task_queue), + Arc::clone(&db), + config.storage.clone(), + ); + tokio::spawn(async move { + album_import_worker.start().await; + }); // 初始化相册内存缓存 let albums_cache = Arc::new(tokio::sync::RwLock::new(AlbumsCache::new())); @@ -128,6 +156,7 @@ async fn main() -> std::io::Result<()> { .app_data(web::Data::new(Arc::clone(&db))) .app_data(web::Data::new(Arc::clone(&image_processor))) .app_data(web::Data::new(Arc::clone(&albums_cache))) + .app_data(web::Data::new(Arc::clone(&task_queue))) // API路由 .route("/api/albums", web::get().to(routes::get_albums)) .route("/api/search", web::get().to(routes::search_albums)) @@ -142,6 +171,8 @@ async fn main() -> std::io::Result<()> { .route("/api/favorites", web::post().to(routes::add_favorite)) .route("/api/favorites/{id}", web::delete().to(routes::remove_favorite)) .route("/api/favorites/image/{id}", web::delete().to(routes::remove_favorite_by_image)) + // 队列状态API路由 + .route("/api/queue/stats", web::get().to(routes::get_queue_stats)) // 静态文件服务 - 启用缓存 .service(afs::Files::new("/assets", "./assets") .show_files_listing() diff --git a/src/models.rs b/src/models.rs index aa67973..5c650ab 100644 --- a/src/models.rs +++ b/src/models.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use tracing::info; use uuid::Uuid; -use tracing::{info, error, warn, debug}; #[derive(Debug, Serialize, Deserialize)] pub struct Album { @@ -94,7 +94,11 @@ pub struct FavoriteStatus { pub is_favorited: bool, pub favorite_id: Option, } - +#[derive(Debug, Deserialize)] +pub struct AlbumIndex { + pub title: String, + pub images: Vec, +} // 内存缓存结构 #[derive(Debug, Clone)] pub struct AlbumsCache { @@ -121,43 +125,47 @@ impl AlbumsCache { pub fn update(&mut self, albums: Vec) { self.albums = albums.clone(); self.last_updated = chrono::Utc::now(); - + // 构建搜索索引 self.search_index.clear(); for album in &self.albums { let title_lower = album.title.to_lowercase(); - + // 为每个单词创建索引 for word in title_lower.split_whitespace() { - if word.len() >= 2 { // 只索引长度>=2的单词 + if word.len() >= 2 { + // 只索引长度>=2的单词 self.search_index .entry(word.to_string()) .or_insert_with(Vec::new) .push(album.clone()); } } - + // 为完整标题创建索引 self.search_index .entry(title_lower.clone()) .or_insert_with(Vec::new) .push(album.clone()); } - - info!("相册缓存已更新，共 {} 个相册，{} 个搜索索引项", - self.albums.len(), self.search_index.len()); + + info!( + "相册缓存已更新，共 {} 个相册，{} 个搜索索引项", + self.albums.len(), + self.search_index.len() + ); } pub fn search(&self, query: &str) -> Vec { let query_lower = query.to_lowercase().trim().to_string(); - + if query_lower.is_empty() { return Vec::new(); } - + let mut results = Vec::new(); let mut seen_ids = std::collections::HashSet::new(); - + // 1. 搜索完整匹配（最高优先级） if let Some(matches) = self.search_index.get(&query_lower) { for album in matches { @@ -166,10 +174,11 @@ impl AlbumsCache { } } } - + // 2. 搜索单词匹配 for word in query_lower.split_whitespace() { - if word.len() >= 2 { // 只搜索长度>=2的单词 + if word.len() >= 2 { + // 只搜索长度>=2的单词 if let Some(matches) = self.search_index.get(word) { for album in matches { if seen_ids.insert(album.id) { @@ -179,15 +188,17 @@ impl AlbumsCache { } } } - + // 3. 模糊搜索：在相册标题和描述中搜索包含查询词的结果 if results.is_empty() { for album in &self.albums { let title_lower = album.title.to_lowercase(); - let description_lower = album.description.as_ref() + let description_lower = album + .description + .as_ref() .map(|d| d.to_lowercase()) .unwrap_or_default(); - + // 检查标题或描述是否包含搜索词 if title_lower.contains(&query_lower) || description_lower.contains(&query_lower) { if seen_ids.insert(album.id) { @@ -196,23 +207,24 @@ impl AlbumsCache { } } } - + // 4. 如果还没有结果，尝试部分匹配（更宽松的搜索） if results.is_empty() && query_lower.len() >= 2 { for album in &self.albums { let title_lower = album.title.to_lowercase(); - + // 检查标题是否包含搜索词的任何部分 - if query_lower.split_whitespace().any(|word| - word.len() >= 2 && title_lower.contains(word) - ) { + if query_lower + .split_whitespace() + .any(|word| word.len() >= 2 && title_lower.contains(word)) + { if seen_ids.insert(album.id) { results.push(album.clone()); } } } } - + results } } \ No newline at end of file diff --git a/src/queue.rs b/src/queue.rs new file mode 100644 index 0000000..23e8079 --- /dev/null +++ b/src/queue.rs @@ -0,0 +1,723 @@ +use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; +use tokio::sync::{Mutex, Notify}; +use tokio::time::sleep; +use tracing::{info, warn, error, debug}; +use uuid::Uuid; +use image::GenericImageView; +use crate::models::{AlbumIndex, CreateAlbum, CreateImage}; + +/// 任务类型 +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum TaskType { + ThumbnailGeneration, // 缩略图生成 + AlbumImport, // 相册导入 +} + +/// 任务优先级 +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub enum TaskPriority { + High = 0, // 前端请求的缩略图 + Medium = 1, // 相册导入 + Low = 2, // 后台质量升级 +} + +/// 通用任务结构 +#[derive(Debug, Clone)] +pub struct Task { + pub id: Uuid, + pub task_type: TaskType, + pub priority: TaskPriority, + pub created_at: Instant, + pub data: TaskData, +} + +/// 任务数据 +#[derive(Debug, Clone)] +pub enum TaskData { + ThumbnailGeneration { + image_id: Uuid, + original_data: Vec, + }, + AlbumImport { + album_path: String, + album_data: Option>, + }, +} + +/// 队列统计信息 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct QueueStats { + pub pending_tasks: usize, + pub active_tasks: usize, + pub completed_tasks: usize, + pub failed_tasks: usize, + pub queue_size_limit: usize, + pub is_paused: bool, + pub task_type_stats: Vec, + // HTTP请求频率信息 + pub request_frequency: f64, + pub request_count: u64, +} + +/// 任务类型统计 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskTypeStats { + pub task_type: TaskType, + pub pending: usize, + pub completed: usize, + pub failed: usize, +} + +/// 智能任务队列 +pub struct TaskQueue { + queue: Mutex>, + active_tasks: AtomicUsize, + completed_tasks: AtomicUsize, + failed_tasks: AtomicUsize, + max_concurrent_tasks: usize, + queue_size_limit: usize, + is_paused: AtomicBool, + notify: Notify, + last_activity: Mutex, + // HTTP请求频率监控 + request_count: AtomicU64, + last_request_time: Mutex, + request_window: Duration, + // 配置阈值 + high_request_rate_threshold: f64, + idle_request_rate_threshold: f64, +} + +impl TaskQueue { + pub fn new(max_concurrent_tasks: usize, queue_size_limit: usize, high_request_rate_threshold: f64, idle_request_rate_threshold: f64) -> Self { + Self { + queue: Mutex::new(VecDeque::new()), + active_tasks: AtomicUsize::new(0), + completed_tasks: AtomicUsize::new(0), + failed_tasks: AtomicUsize::new(0), + max_concurrent_tasks, + queue_size_limit, + is_paused: AtomicBool::new(false), + notify: Notify::new(), + last_activity: Mutex::new(Instant::now()), + // HTTP请求频率监控 + request_count: AtomicU64::new(0), + last_request_time: Mutex::new(Instant::now()), + request_window: Duration::from_secs(10), + // 配置阈值 + high_request_rate_threshold, + idle_request_rate_threshold, + } + } + + /// 添加任务到队列 + pub async fn add_task(&self, task: Task) -> Result<(), String> { + let mut queue = self.queue.lock().await; + + // 检查队列是否已满 + if queue.len() >= self.queue_size_limit { + return Err("队列已满，无法添加新任务".to_string()); + } + + // 检查是否已有相同ID的任务 + if queue.iter().any(|t| t.id == task.id) { + return Err("该任务已在队列中".to_string()); + } + + // 根据优先级插入任务（高优先级在前） + let insert_position = queue + .iter() + .position(|t| t.priority > task.priority) + .unwrap_or(queue.len()); + + queue.insert(insert_position, task); + + // 更新最后活动时间 + *self.last_activity.lock().await = Instant::now(); + + // 通知工作线程 + self.notify.notify_one(); + + debug!("任务已添加到队列，当前队列大小: {}", queue.len()); + Ok(()) + } + + /// 从队列获取下一个任务 + pub async fn next_task(&self) -> Option { + // 检查是否暂停 + if self.is_paused.load(Ordering::Acquire) { + return None; + } + + // 检查并发任务数量 + if self.active_tasks.load(Ordering::Acquire) >= self.max_concurrent_tasks { + return None; + } + + let mut queue = self.queue.lock().await; + if let Some(task) = queue.pop_front() { + self.active_tasks.fetch_add(1, Ordering::Release); + Some(task) + } else { + None + } + } + + /// 标记任务完成 + pub fn task_completed(&self, success: bool) { + self.active_tasks.fetch_sub(1, Ordering::Release); + if success { + self.completed_tasks.fetch_add(1, Ordering::Release); + } else { + self.failed_tasks.fetch_add(1, Ordering::Release); + } + } + + /// 暂停队列处理 + pub fn pause(&self) { + self.is_paused.store(true, Ordering::Release); + info!("任务队列已暂停"); + } + + /// 恢复队列处理 + pub fn resume(&self) { + self.is_paused.store(false, Ordering::Release); + self.notify.notify_one(); + info!("任务队列已恢复"); + } + + /// 记录HTTP请求（在每次HTTP请求处理时调用） + pub async fn record_http_request(&self) { + let now = Instant::now(); + let mut last_request_time = self.last_request_time.lock().await; + + // 如果距离上次请求时间超过了窗口时间，重置计数器 + if now.duration_since(*last_request_time) > self.request_window { + self.request_count.store(0, Ordering::Release); + } + + // 增加请求计数 + self.request_count.fetch_add(1, Ordering::Release); + *last_request_time = now; + } + + /// 获取当前请求频率（请求数/秒） + pub fn get_request_frequency(&self) -> f64 { + let request_count = self.request_count.load(Ordering::Acquire) as f64; + let window_seconds = self.request_window.as_secs_f64(); + + if window_seconds > 0.0 { + request_count / window_seconds + } else { + 0.0 + } + } + + /// 检查服务器负载是否过高（基于HTTP请求频率） + pub async fn is_server_busy(&self) -> bool { + let request_frequency = self.get_request_frequency(); + + // 基于HTTP请求频率判断业务繁忙度 + // 高请求频率（超过配置阈值）时认为服务器繁忙 + let is_high_request_rate = request_frequency > self.high_request_rate_threshold; + + // 如果请求频率高，或者并发任务/队列大小超过阈值，认为服务器繁忙 + is_high_request_rate + } + + /// 获取队列统计信息 + pub async fn get_stats(&self) -> QueueStats { + let queue = self.queue.lock().await; + + // 按任务类型统计 + let mut type_stats = std::collections::HashMap::new(); + for task in queue.iter() { + let stats = type_stats + .entry(task.task_type.clone()) + .or_insert((0, 0, 0)); + stats.0 += 1; // pending + } + + let task_type_stats = type_stats + .into_iter() + .map(|(task_type, (pending, completed, failed))| TaskTypeStats { + task_type, + pending, + completed, + failed, + }) + .collect(); + + QueueStats { + pending_tasks: queue.len(), + active_tasks: self.active_tasks.load(Ordering::Acquire), + completed_tasks: self.completed_tasks.load(Ordering::Acquire), + failed_tasks: self.failed_tasks.load(Ordering::Acquire), + queue_size_limit: self.queue_size_limit, + is_paused: self.is_paused.load(Ordering::Acquire), + task_type_stats, + // 添加请求频率信息 + request_frequency: self.get_request_frequency(), + request_count: self.request_count.load(Ordering::Acquire), + } + } + + /// 清理过期的任务（超过一定时间的任务） + pub async fn cleanup_expired_tasks(&self, max_age: Duration) { + let mut queue = self.queue.lock().await; + let now = Instant::now(); + let initial_len = queue.len(); + + queue.retain(|task| now.duration_since(task.created_at) <= max_age); + + let removed_count = initial_len - queue.len(); + if removed_count > 0 { + warn!("清理了 {} 个过期任务", removed_count); + } + } + + /// 等待队列变为空闲状态 + pub async fn wait_for_idle(&self, timeout: Duration) -> bool { + let start = Instant::now(); + + while start.elapsed() < timeout { + sleep(Duration::from_millis(100)).await; + } + + false + } +} + +/// 缩略图生成Worker +pub struct ThumbnailWorker { + queue: Arc, + image_processor: Arc, +} + +impl ThumbnailWorker { + pub fn new( + queue: Arc, + image_processor: Arc, + ) -> Self { + Self { + queue, + image_processor, + } + } + + /// 启动缩略图Worker + pub async fn start(self) { + info!("启动缩略图生成Worker"); + + loop { + // 检查队列中是否有任务 + if let Some(task) = self.queue.next_task().await { + if matches!(task.task_type, TaskType::ThumbnailGeneration) { + debug!("缩略图Worker开始处理任务: {:?}", task.id); + + let success = self.process_thumbnail_task(task).await; + self.queue.task_completed(success); + + // 在处理完每个任务后检查服务器负载 + if self.queue.is_server_busy().await { + info!("服务器负载较高，暂停队列处理"); + self.queue.pause(); + + // 每30秒检查一次服务器状态，直到服务器不忙时恢复队列 + loop { + tokio::time::sleep(Duration::from_secs(30)).await; + if !self.queue.is_server_busy().await { + info!("服务器负载恢复正常，恢复队列处理"); + self.queue.resume(); + break; + } + info!("服务器仍然繁忙，继续等待..."); + } + } + } else { + // 如果不是缩略图任务，放回队列 + let mut queue = self.queue.queue.lock().await; + queue.push_front(task); + } + } else { + // 队列为空，等待新任务通知 + self.queue.notify.notified().await; + } + + // 短暂休眠避免CPU占用过高 + sleep(Duration::from_millis(100)).await; + } + } + + /// 处理缩略图任务 + async fn process_thumbnail_task(&self, task: Task) -> bool { + if let TaskData::ThumbnailGeneration { + image_id, + original_data, + } = task.data + { + // 使用具体的错误处理，避免 Send trait 问题 + let result = self + .image_processor + .generate_thumbnail_from_memory(&original_data, &image_id.to_string()) + .await; + + match result { + Ok(thumbnail_data) => { + // 保存缩略图到文件 + let thumbnail_path = self + .image_processor + .config + .thumbnails_dir + .join(image_id.to_string()); + + match tokio::fs::write(&thumbnail_path, &thumbnail_data).await { + Ok(_) => { + info!("✅ 缩略图生成成功: {:?}", image_id); + true + } + Err(e) => { + warn!("保存缩略图失败: {:?} - {}", image_id, e); + false + } + } + } + Err(e) => { + warn!("缩略图生成失败: {:?} - {:?}", image_id, e); + false + } + } + } else { + warn!("任务类型不匹配，期望缩略图生成任务"); + false + } + } +} + +/// 相册导入Worker +pub struct AlbumImportWorker { + queue: Arc, + db: Arc, + config: crate::config::StorageConfig, +} + +impl AlbumImportWorker { + pub fn new( + queue: Arc, + db: Arc, + config: crate::config::StorageConfig, + ) -> Self { + Self { queue, db, config } + } + + /// 启动相册导入Worker + pub async fn start(self) { + info!("启动相册导入Worker"); + + loop { + // 检查队列中是否有任务 + if let Some(task) = self.queue.next_task().await { + if matches!(task.task_type, TaskType::AlbumImport) { + debug!("相册导入Worker开始处理任务: {:?}", task.id); + + let success = self.process_album_import_task(task).await; + self.queue.task_completed(success); + + // 在处理完每个任务后检查服务器负载 + if self.queue.is_server_busy().await { + info!("服务器负载较高，暂停队列处理"); + self.queue.pause(); + + // 每30秒检查一次服务器状态，直到服务器不忙时恢复队列 + loop { + tokio::time::sleep(Duration::from_secs(30)).await; + if !self.queue.is_server_busy().await { + info!("服务器负载恢复正常，恢复队列处理"); + self.queue.resume(); + break; + } + info!("服务器仍然繁忙，继续等待..."); + } + } + } else { + // 如果不是相册导入任务，放回队列并减少active_tasks计数器 + let mut queue = self.queue.queue.lock().await; + queue.push_front(task); + // 手动减少active_tasks计数器，因为任务没有被处理 + self.queue.active_tasks.fetch_sub(1, Ordering::Release); + } + } else { + // 队列为空，等待新任务通知 + self.queue.notify.notified().await; + } + + // 短暂休眠避免CPU占用过高 + sleep(Duration::from_millis(100)).await; + } + } + + /// 处理相册导入任务 + async fn process_album_import_task(&self, task: Task) -> bool { + if let TaskData::AlbumImport { + album_path, + album_data: _, + } = task.data + { + info!("开始处理相册导入任务: {}", album_path); + + // 将相对路径与images_dir组合成完整路径 + let full_path = std::path::Path::new(&self.config.images_dir).join(&album_path); + + // 检查相册文件夹是否存在 + if !full_path.exists() { + error!("相册文件夹不存在: {}", full_path.display()); + return false; + } + + // 导入单个相册 + match self.import_single_album(&album_path, &full_path).await { + Ok(count) => { + info!("相册导入成功: {} (导入 {} 张图片)", album_path, count); + true + } + Err(e) => { + error!("相册导入失败: {} - {}", album_path, e); + false + } + } + } else { + warn!("任务类型不匹配，期望相册导入任务"); + false + } + } + + /// 导入单个相册（按照import.rs中的逻辑实现） + async fn import_single_album(&self, folder_name: &str, path: &std::path::Path) -> Result> { + info!("开始导入相册: {}", folder_name); + + // 读取index.json文件 + let index_file_path = path.join("index.json"); + let album_title = if index_file_path.exists() { + match std::fs::read_to_string(&index_file_path) { + Ok(content) => { + match serde_json::from_str::(&content) { + Ok(index) => { + info!("从index.json读取相册标题: {}", index.title); + index.title + } + Err(e) => { + warn!("解析index.json失败: {}, 使用文件夹名作为标题", e); + folder_name.to_string() + } + } + } + Err(e) => { + warn!("读取index.json失败: {}, 使用文件夹名作为标题", e); + folder_name.to_string() + } + } + } else { + info!("index.json不存在，使用文件夹名作为标题"); + folder_name.to_string() + }; + + // 创建新相册 + let new_album = CreateAlbum { + title: album_title.clone(), + document: folder_name.to_string(), + description: Some(format!("从文件夹 {} 自动导入的相册", folder_name)), + }; + + let album_id = self.db.create_album(new_album).await?; + info!("创建相册: {} (ID: {})", album_title, album_id); + + // 读取图片URL列表（如果有index.json） + let image_urls = if index_file_path.exists() { + match std::fs::read_to_string(&index_file_path) { + Ok(content) => { + match serde_json::from_str::(&content) { + Ok(index) => { + info!("从index.json读取到 {} 个图片URL", index.images.len()); + Some(index.images) + } + Err(_) => None + } + } + Err(_) => None + } + } else { + None + }; + + // 快速扫描图片文件 + let image_files = self.scan_image_files(path, folder_name)?; + + if image_files.is_empty() { + info!("相册 {} 中没有找到图片文件", folder_name); + return Ok(0); + } + + info!("发现 {} 张图片，开始处理...", image_files.len()); + + // 并行处理图片信息获取 + let mut image_tasks = Vec::new(); + + for (index, (filename, image_path)) in image_files.iter().enumerate() { + let image_url = image_urls.as_ref().and_then(|urls| { + if urls.len() > index { + Some(urls[index].clone()) + } else { + None + } + }); + + let filename_clone = filename.clone(); + let image_path_clone = image_path.clone(); + let album_id_clone = album_id; + let folder_name_clone = folder_name.to_string(); + + let task = tokio::task::spawn_blocking(move || { + // 在阻塞线程中处理图片信息 + let (file_size, width, height) = match AlbumImportWorker::get_image_info_static(&image_path_clone) { + Ok(info) => info, + Err(e) => { + eprintln!("获取图片信息失败: {} - {}", image_path_clone.display(), e); + (0, 0, 0) + } + }; + + CreateImage { + album_id: album_id_clone, + filename: format!("{}/{}", folder_name_clone, filename_clone), + original_url: image_url, + title: Some(format!("图片 {}", index + 1)), + description: Some(format!("从文件夹 {} 导入的图片", folder_name_clone)), + file_size, + width, + height, + mime_type: match image_path_clone.extension().and_then(|s| s.to_str()).unwrap_or("jpg").to_lowercase().as_str() { + "jpg" | "jpeg" => "image/jpeg".to_string(), + "png" => "image/png".to_string(), + "gif" => "image/gif".to_string(), + "webp" => "image/webp".to_string(), + "bmp" => "image/bmp".to_string(), + _ => "image/jpeg".to_string(), + }, + } + }); + + image_tasks.push(task); + } + + // 收集所有图片信息 + let mut images_to_insert = Vec::new(); + for task in image_tasks { + match task.await { + Ok(image) => images_to_insert.push(image), + Err(e) => error!("图片处理任务失败: {}", e), + } + } + + // 批量插入图片 + if !images_to_insert.is_empty() { + match self.db.add_images(images_to_insert).await { + Ok(ids) => { + info!("成功导入 {} 张图片到相册 {}", ids.len(), folder_name); + + // 更新相册图片数量 + if let Err(e) = self.db.update_album_image_count(album_id).await { + error!("更新相册图片数量失败: {}", e); + } + + Ok(ids.len()) + } + Err(e) => { + error!("批量插入图片失败: {}", e); + Ok(0) + } + } + } else { + info!("没有图片需要插入到相册 {}", folder_name); + Ok(0) + } + } + + /// 扫描图片文件（从import.rs迁移） + fn scan_image_files(&self, path: &std::path::Path, _folder_name: &str) -> Result, Box> { + let mut image_files = Vec::new(); + + for entry in std::fs::read_dir(path)? { + let entry = entry?; + let file_path = entry.path(); + + if file_path.is_file() { + if let Some(extension) = file_path.extension().and_then(|s| s.to_str()) { + let ext_lower = extension.to_lowercase(); + if ext_lower == "jpg" || ext_lower == "jpeg" || ext_lower == "png" || + ext_lower == "gif" || ext_lower == "webp" || ext_lower == "bmp" { + + if let Some(filename) = file_path.file_name().and_then(|s| s.to_str()) { + image_files.push((filename.to_string(), file_path)); + } + } + } + } + } + + Ok(image_files) + } + + + /// 静态方法：获取图片信息（用于spawn_blocking） + fn get_image_info_static(image_path: &std::path::Path) -> Result<(i64, i32, i32), Box> { + let metadata = std::fs::metadata(image_path)?; + let file_size = metadata.len() as i64; + + // 使用image库获取图片尺寸 + let img = image::open(image_path)?; + let (width, height) = img.dimensions(); + + Ok((file_size, width as i32, height as i32)) + } +} + +/// 任务工厂 - 用于创建不同类型的任务 +pub struct TaskFactory; + +impl TaskFactory { + /// 创建缩略图生成任务 + pub fn create_thumbnail_task( + image_id: Uuid, + original_data: Vec, + priority: TaskPriority, + ) -> Task { + Task { + id: Uuid::new_v4(), + task_type: TaskType::ThumbnailGeneration, + priority, + created_at: Instant::now(), + data: TaskData::ThumbnailGeneration { + image_id, + original_data, + }, + } + } + + /// 创建相册导入任务 + pub fn create_album_import_task(album_path: String, album_data: Option>) -> Task { + Task { + id: Uuid::new_v4(), + task_type: TaskType::AlbumImport, + priority: TaskPriority::Medium, + created_at: Instant::now(), + data: TaskData::AlbumImport { + album_path, + album_data, + }, + } + } +} \ No newline at end of file diff --git a/src/routes.rs b/src/routes.rs index e93ffae..621bb8b 100644 --- a/src/routes.rs +++ b/src/routes.rs @@ -7,6 +7,16 @@ use std::sync::Arc; use tokio::sync::RwLock; use tracing::error; use uuid::Uuid; +use crate::queue; + +// 记录HTTP请求频率的辅助函数 +async fn log_request_frequency(method: &str, path: &str, client_ip: Option<&str>) { + // 这里可以添加请求频率记录逻辑 + // 例如：记录到数据库、内存缓存或日志文件 + let ip = client_ip.unwrap_or("unknown"); + println!("Request: {} {} from {}", method, path, ip); +} + // 统一错误处理宏 macro_rules! handle_db_error { ($result:expr, $error_msg:expr) => { @@ -61,7 +71,8 @@ async fn check_and_upgrade_thumbnail_quality( config: &crate::config::StorageConfig, image: &Image, original_path: &std::path::Path, -) -> Result<(), Box> { + task_queue: Option>, +) -> Result<(), Box> { let thumbnail_path = config.thumbnails_dir.join(image.id.to_string()); // 检查原始图片是否存在 @@ -87,22 +98,35 @@ async fn check_and_upgrade_thumbnail_quality( } }; - // 生成高质量缩略图 - let image_processor = ImageProcessor::new(config.clone()); - match image_processor - .generate_thumbnail_from_memory(&original_data, &image.id.to_string()) - .await - { - Ok(high_quality_data) => { - // 保存高质量缩略图 - if let Err(e) = fs::write(&thumbnail_path, &high_quality_data) { - eprintln!("保存高质量缩略图失败: {}", e); - } else { - println!("✅ 缩略图已升级为高质量版本 (ID: {})", image.id); - } + // 如果有队列系统，使用队列处理质量升级 + if let Some(queue) = task_queue { + let task = queue::TaskFactory::create_thumbnail_task( + image.id, + original_data, + queue::TaskPriority::Low, // 质量升级，低优先级 + ); + + if let Err(e) = queue.add_task(task).await { + eprintln!("添加缩略图质量升级任务失败: {}", e); } - Err(e) => { - eprintln!("生成高质量缩略图失败: {}", e); + } else { + // 如果没有队列系统，直接生成高质量缩略图 + let image_processor = ImageProcessor::new(config.clone()); + match image_processor + .generate_thumbnail_from_memory(&original_data, &image.id.to_string()) + .await + { + Ok(high_quality_data) => { + // 保存高质量缩略图 + if let Err(e) = fs::write(&thumbnail_path, &high_quality_data) { + eprintln!("保存高质量缩略图失败: {}", e); + } else { + println!("✅ 缩略图已升级为高质量版本 (ID: {})", image.id); + } + } + Err(e) => { + eprintln!("生成高质量缩略图失败: {:?}", e); + } } } } @@ -184,6 +208,17 @@ pub async fn remove_favorite( } } +// 获取任务队列状态 +pub async fn get_queue_stats( + task_queue: web::Data>, +) -> impl Responder { + // 记录HTTP请求频率 + task_queue.record_http_request().await; + + let stats = task_queue.get_stats().await; + HttpResponse::Ok().json(stats) +} + // 通过图片ID取消收藏 pub async fn remove_favorite_by_image( db: web::Data>, @@ -418,6 +453,7 @@ pub async fn create_album( pub async fn get_thumbnail( db: web::Data>, image_processor: web::Data>, + task_queue: web::Data>, path: web::Path, ) -> impl Responder { let image_id = match Uuid::parse_str(&path.into_inner()) { @@ -448,11 +484,13 @@ pub async fn get_thumbnail( let config_clone = image_processor.config.clone(); let image_clone = image.clone(); let original_path_clone = original_path.clone(); + let queue_clone = task_queue.get_ref().clone(); tokio::spawn(async move { if let Err(e) = check_and_upgrade_thumbnail_quality( &config_clone, &image_clone, &original_path_clone, + Some(queue_clone), ) .await { @@ -483,28 +521,18 @@ pub async fn get_thumbnail( .await { Ok(thumbnail_data) => { - // 异步保存高质量缩略图到文件（不阻塞响应） - let config_clone = image_processor.config.clone(); - let filename_clone = image.id.clone(); - let original_data_clone = data.clone(); + // 将高质量缩略图生成任务添加到队列 + let task = queue::TaskFactory::create_thumbnail_task( + image.id, + data.clone(), + queue::TaskPriority::High, // 前端请求，高优先级 + ); + + // 异步添加任务到队列 + let queue_clone = task_queue.clone(); tokio::spawn(async move { - let thumbnail_path = config_clone - .thumbnails_dir - .join(filename_clone.to_string()); - - // 生成高质量缩略图并保存 - if let Ok(high_quality_data) = ImageProcessor::new(config_clone) - .generate_thumbnail_from_memory( - &original_data_clone, - &filename_clone.to_string(), - ) - .await - { - if let Err(e) = - fs::write(&thumbnail_path, &high_quality_data) - { - eprintln!("保存高质量缩略图失败: {}", e); - } + if let Err(e) = queue_clone.add_task(task).await { + eprintln!("添加缩略图生成任务失败: {}", e); } }); @@ -516,7 +544,7 @@ pub async fn get_thumbnail( .body(thumbnail_data) } Err(e) => { - eprintln!("快速缩略图生成失败: {}", e); + eprintln!("快速缩略图生成失败: {:?}", e); // 如果快速生成失败，返回原始图片的小尺寸版本 let small_data = image_processor.resize_image_to_max(&data, 800).await; @@ -549,4 +577,4 @@ pub async fn get_thumbnail( })) } } -} +} \ No newline at end of file