diff --git a/src/database.rs b/src/database.rs index 2d1e248..b797f2c 100644 --- a/src/database.rs +++ b/src/database.rs @@ -1,7 +1,9 @@ -use sqlx::{SqlitePool, sqlite::SqlitePoolOptions, Row}; +use sqlx::{SqlitePool, sqlite::SqlitePoolOptions, Row, Error as SqlxError}; use crate::config::Config; use crate::models::*; use uuid::Uuid; +use std::time::Duration; +use tokio::time::sleep; pub struct Database { pool: SqlitePool, @@ -31,10 +33,24 @@ impl Database { } let pool = SqlitePoolOptions::new() - .max_connections(5) + .max_connections(10) // 增加连接数 + .min_connections(2) // 保持最小连接数 + .acquire_timeout(std::time::Duration::from_secs(30)) // 增加获取连接超时时间 + .idle_timeout(std::time::Duration::from_secs(300)) // 增加空闲超时时间 + .max_lifetime(std::time::Duration::from_secs(1800)) // 增加最大生命周期 .connect(url) .await?; + // 在迁移之前设置SQLite性能优化参数 + // 这些PRAGMA语句必须在事务外执行 + sqlx::query("PRAGMA journal_mode=WAL;").execute(&pool).await?; + sqlx::query("PRAGMA wal_autocheckpoint=4000;").execute(&pool).await?; + sqlx::query("PRAGMA synchronous=NORMAL;").execute(&pool).await?; + sqlx::query("PRAGMA cache_size=-64000;").execute(&pool).await?; + sqlx::query("PRAGMA mmap_size=268435456;").execute(&pool).await?; + sqlx::query("PRAGMA temp_store=MEMORY;").execute(&pool).await?; + sqlx::query("PRAGMA foreign_keys=ON;").execute(&pool).await?; + // 创建表 sqlx::migrate!("./migrations").run(&pool).await?; @@ -42,6 +58,53 @@ impl Database { } + /// 带重试机制的数据库操作执行器 + async fn execute_with_retry(&self, operation: F, operation_name: &str) -> Result + where + F: Fn() -> std::pin::Pin> + Send>>, + { + const MAX_RETRIES: u32 = 3; + const BASE_DELAY_MS: u64 = 100; + + for attempt in 0..=MAX_RETRIES { + match operation().await { + Ok(result) => return Ok(result), + Err(e) => { + // 检查是否是死锁或超时错误 + if self.is_retryable_error(&e) && attempt < MAX_RETRIES { + let delay_ms = BASE_DELAY_MS * 2u64.pow(attempt); + eprintln!("⚠️ {} 操作失败 (尝试 {}): {}, 等待 {}ms 后重试", + operation_name, attempt + 1, e, delay_ms); + sleep(Duration::from_millis(delay_ms)).await; + } else { + eprintln!("❌ {} 操作失败 (尝试 {}): {}", operation_name, attempt + 1, e); + return Err(e); + } + } + } + } + + unreachable!() + } + + /// 判断错误是否可重试 + fn is_retryable_error(&self, error: &SqlxError) -> bool { + match error { + SqlxError::Database(db_err) => { + // SQLite 死锁错误码 + if let Some(code) = db_err.code() { + code == "5" || // SQLITE_BUSY + code == "6" || // SQLITE_LOCKED + code == "261" // SQLITE_BUSY_SNAPSHOT + } else { + false + } + } + SqlxError::PoolTimedOut => true, + _ => false, + } + } + // 相册操作 pub async fn get_all_albums(&self) -> Result, sqlx::Error> { let albums = sqlx::query( @@ -299,6 +362,7 @@ impl Database { return Ok(Vec::new()); } + // 由于生命周期问题,简化实现,不使用重试机制 let now = chrono::Utc::now().to_rfc3339(); let mut ids = Vec::new(); diff --git a/src/import.rs b/src/import.rs index f155f9d..f112b15 100644 --- a/src/import.rs +++ b/src/import.rs @@ -57,52 +57,32 @@ pub async fn import_existing_data(config: &Config, db: Arc) -> Result< return Ok(()); } - // 第二阶段:并行处理相册导入 - let mut tasks = Vec::new(); - - // 限制并发数量,避免内存爆炸 - let max_concurrent = std::thread::available_parallelism()?.get().min(4); - let semaphore = Arc::new(tokio::sync::Semaphore::new(max_concurrent)); - - for (folder_name, path) in album_folders { - let db_clone = Arc::clone(&db); - let semaphore_clone = Arc::clone(&semaphore); - let folder_name_clone = folder_name.clone(); - let path_clone = path.clone(); - - let task = task::spawn(async move { - let _permit = semaphore_clone.acquire().await.unwrap(); - match import_single_album(&db_clone, &folder_name_clone, &path_clone).await { - Ok(count) => { - println!("✅ 相册 {} 导入完成,共导入 {} 张图片", folder_name_clone, count); - count - } - Err(e) => { - eprintln!("❌ 相册 {} 导入失败: {}", folder_name_clone, e); - 0 - } - } - }); - - tasks.push(task); - } - - // 等待所有任务完成 + // 第二阶段:串行处理相册导入,避免数据库死锁 + // 对于SQLite,高并发写入容易导致死锁,改为串行处理 let mut total_imported = 0; let mut successful_albums = 0; - for task in tasks { - match task.await { + 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!("任务执行失败: {}", e); + eprintln!("❌ 相册 {} 导入失败: {}", folder_name, e); + // 继续处理下一个相册,不中断整个导入过程 } } + + // 在每个相册导入后添加短暂延迟,减少数据库压力 + tokio::time::sleep(std::time::Duration::from_millis(100)).await; } println!("🎉 导入完成!成功导入 {} 个相册,共 {} 张图片", successful_albums, total_imported);