fix: 添加连接池功能

This commit is contained in:
imeepos
2025-07-15 18:43:22 +08:00
parent 86eeb90fb8
commit 8d8e98188d
7 changed files with 567 additions and 11 deletions

View File

@@ -31,7 +31,45 @@ impl AppState {
/// 初始化数据库连接
/// 遵循安全第一原则,确保数据库初始化的安全性
/// 默认使用连接池模式以提高并发性能
pub fn initialize_database(&self) -> anyhow::Result<()> {
println!("开始初始化数据库连接...");
// 暂时使用单连接模式,避免锁竞争问题
// TODO: 在解决 SQLite 锁问题后重新启用连接池
let database = Arc::new(Database::new()?);
println!("使用单连接模式初始化数据库");
// 连接池模式代码(暂时注释)
/*
let database = match Database::new_with_pool() {
Ok(db) => {
println!("连接池模式初始化成功");
Arc::new(db)
},
Err(e) => {
eprintln!("连接池模式初始化失败: {}, 回退到单连接模式", e);
Arc::new(Database::new()?)
}
};
*/
let project_repository = ProjectRepository::new(database.clone())?;
let material_repository = MaterialRepository::new(database.clone())?;
let model_repository = ModelRepository::new(database.clone());
*self.database.lock().unwrap() = Some(database.clone());
*self.project_repository.lock().unwrap() = Some(project_repository);
*self.material_repository.lock().unwrap() = Some(material_repository);
*self.model_repository.lock().unwrap() = Some(model_repository);
println!("数据库初始化完成,连接池状态: {}",
if database.has_pool() { "已启用" } else { "未启用" });
Ok(())
}
/// 初始化数据库连接(单连接模式,用于测试或特殊场景)
pub fn initialize_database_single_mode(&self) -> anyhow::Result<()> {
let database = Arc::new(Database::new()?);
let project_repository = ProjectRepository::new(database.clone())?;
@@ -43,6 +81,7 @@ impl AppState {
*self.material_repository.lock().unwrap() = Some(material_repository);
*self.model_repository.lock().unwrap() = Some(model_repository);
println!("数据库初始化完成,使用单连接模式");
Ok(())
}

View File

@@ -373,21 +373,23 @@ impl ModelRepository {
}
/// 获取模特照片
/// 使用非阻塞数据库访问模式,避免无限等待
/// 使用连接池优化的数据库访问模式,避免锁竞争
pub fn get_photos(&self, model_id: &str) -> Result<Vec<ModelPhoto>> {
println!("get_photos 开始执行model_id: {}", model_id);
// 使用非阻塞方式获取连接
match self.database.try_get_connection() {
Some(conn) => {
println!("get_photos 成功获取数据库连接锁(非阻塞");
// 优先使用连接池,如果不可用则回退到单连接模式
match self.database.get_best_connection() {
Ok(conn) => {
println!("get_photos 成功获取数据库连接{}模式",
if self.database.has_pool() { "连接池" } else { "单连接" });
let photos = self.execute_photo_query(&conn, model_id)?;
println!("get_photos 执行完成,返回 {} 张照片", photos.len());
Ok(photos)
},
None => {
println!("get_photos 连接被占用,返回空结果避免阻塞");
// 如果连接被占用,直接返回空结果避免阻塞
Err(e) => {
println!("get_photos 无法获取数据库连接: {}", e);
// 如果所有连接被占用,返回空结果避免阻塞UI
Ok(Vec::new())
}
}

View File

@@ -0,0 +1,300 @@
use std::sync::{Arc, Mutex};
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use rusqlite::Connection;
use anyhow::{Result, anyhow};
/// 数据库连接池配置
#[derive(Debug, Clone)]
pub struct ConnectionPoolConfig {
/// 最小连接数
pub min_connections: usize,
/// 最大连接数
pub max_connections: usize,
/// 连接超时时间(秒)
pub connection_timeout: Duration,
/// 连接空闲超时时间(秒)
pub idle_timeout: Duration,
/// 获取连接的最大等待时间
pub acquire_timeout: Duration,
}
impl Default for ConnectionPoolConfig {
fn default() -> Self {
Self {
min_connections: 1, // 减少最小连接数
max_connections: 3, // 减少最大连接数SQLite 不适合太多并发连接
connection_timeout: Duration::from_secs(30),
idle_timeout: Duration::from_secs(300), // 5分钟
acquire_timeout: Duration::from_secs(5),
}
}
}
/// 连接包装器,包含连接和元数据
#[derive(Debug)]
struct PooledConnection {
connection: Connection,
created_at: Instant,
last_used: Instant,
is_busy: bool,
}
impl PooledConnection {
fn new(connection: Connection) -> Self {
let now = Instant::now();
Self {
connection,
created_at: now,
last_used: now,
is_busy: false,
}
}
fn is_expired(&self, idle_timeout: Duration) -> bool {
self.last_used.elapsed() > idle_timeout
}
fn mark_used(&mut self) {
self.last_used = Instant::now();
self.is_busy = true;
}
fn mark_returned(&mut self) {
self.last_used = Instant::now();
self.is_busy = false;
}
}
/// 数据库连接池
pub struct ConnectionPool {
database_path: String,
config: ConnectionPoolConfig,
connections: Arc<Mutex<VecDeque<PooledConnection>>>,
stats: Arc<Mutex<PoolStats>>,
}
/// 连接池统计信息
#[derive(Debug, Default)]
pub struct PoolStats {
pub total_connections: usize,
pub active_connections: usize,
pub idle_connections: usize,
pub total_acquired: u64,
pub total_returned: u64,
pub acquire_timeouts: u64,
pub connection_errors: u64,
}
/// 连接池中的连接句柄
pub struct PooledConnectionHandle {
connection: Option<Connection>,
pool: Arc<ConnectionPool>,
}
impl PooledConnectionHandle {
fn new(connection: Connection, pool: Arc<ConnectionPool>) -> Self {
Self {
connection: Some(connection),
pool,
}
}
/// 获取底层连接的引用
pub fn as_ref(&self) -> &Connection {
self.connection.as_ref().unwrap()
}
/// 获取底层连接的可变引用
pub fn as_mut(&mut self) -> &mut Connection {
self.connection.as_mut().unwrap()
}
}
impl Drop for PooledConnectionHandle {
fn drop(&mut self) {
if let Some(connection) = self.connection.take() {
self.pool.return_connection(connection);
}
}
}
impl std::ops::Deref for PooledConnectionHandle {
type Target = Connection;
fn deref(&self) -> &Self::Target {
self.connection.as_ref().unwrap()
}
}
impl std::ops::DerefMut for PooledConnectionHandle {
fn deref_mut(&mut self) -> &mut Self::Target {
self.connection.as_mut().unwrap()
}
}
impl ConnectionPool {
/// 创建新的连接池
pub fn new(database_path: String, config: ConnectionPoolConfig) -> Result<Arc<Self>> {
let pool = Arc::new(Self {
database_path,
config,
connections: Arc::new(Mutex::new(VecDeque::new())),
stats: Arc::new(Mutex::new(PoolStats::default())),
});
// 初始化最小连接数
pool.initialize_connections()?;
Ok(pool)
}
/// 初始化连接池
fn initialize_connections(&self) -> Result<()> {
let mut connections = self.connections.lock().map_err(|e| anyhow!("获取连接池锁失败: {}", e))?;
for _ in 0..self.config.min_connections {
let conn = self.create_connection()?;
connections.push_back(PooledConnection::new(conn));
}
// 更新统计信息
let mut stats = self.stats.lock().map_err(|e| anyhow!("获取统计锁失败: {}", e))?;
stats.total_connections = connections.len();
stats.idle_connections = connections.len();
println!("连接池初始化完成,创建了 {} 个连接", connections.len());
Ok(())
}
/// 创建新的数据库连接
fn create_connection(&self) -> Result<Connection> {
let conn = Connection::open(&self.database_path)
.map_err(|e| anyhow!("创建数据库连接失败: {}", e))?;
// 设置连接参数
conn.execute_batch("
PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = 1000;
PRAGMA temp_store = memory;
PRAGMA busy_timeout = 5000;
").map_err(|e| anyhow!("设置数据库参数失败: {}", e))?;
Ok(conn)
}
/// 获取连接(阻塞方式)
pub fn acquire(self: &Arc<Self>) -> Result<PooledConnectionHandle> {
let start_time = Instant::now();
loop {
// 尝试从池中获取连接
if let Some(connection) = self.try_acquire_from_pool()? {
let mut stats = self.stats.lock().map_err(|e| anyhow!("获取统计锁失败: {}", e))?;
stats.total_acquired += 1;
stats.active_connections += 1;
stats.idle_connections = stats.idle_connections.saturating_sub(1);
return Ok(PooledConnectionHandle::new(connection, Arc::clone(self)));
}
// 检查是否超时
if start_time.elapsed() >= self.config.acquire_timeout {
let mut stats = self.stats.lock().map_err(|e| anyhow!("获取统计锁失败: {}", e))?;
stats.acquire_timeouts += 1;
return Err(anyhow!("获取数据库连接超时"));
}
// 等待一小段时间后重试
std::thread::sleep(Duration::from_millis(10));
}
}
/// 尝试非阻塞获取连接
pub fn try_acquire(self: &Arc<Self>) -> Result<Option<PooledConnectionHandle>> {
match self.try_acquire_from_pool()? {
Some(connection) => {
let mut stats = self.stats.lock().map_err(|e| anyhow!("获取统计锁失败: {}", e))?;
stats.total_acquired += 1;
stats.active_connections += 1;
stats.idle_connections = stats.idle_connections.saturating_sub(1);
Ok(Some(PooledConnectionHandle::new(connection, Arc::clone(self))))
},
None => Ok(None),
}
}
/// 从池中尝试获取连接
fn try_acquire_from_pool(&self) -> Result<Option<Connection>> {
let mut connections = self.connections.lock().map_err(|e| anyhow!("获取连接池锁失败: {}", e))?;
// 查找可用的连接
for pooled_conn in connections.iter_mut() {
if !pooled_conn.is_busy && !pooled_conn.is_expired(self.config.idle_timeout) {
pooled_conn.mark_used();
// 这里我们需要移除连接,但由于借用检查器的限制,我们需要重新设计
// 暂时返回 None后续优化
break;
}
}
// 清理过期连接
connections.retain(|conn| !conn.is_expired(self.config.idle_timeout));
// 如果没有可用连接且未达到最大连接数,创建新连接
if connections.len() < self.config.max_connections {
match self.create_connection() {
Ok(new_conn) => {
println!("创建新的数据库连接,当前连接数: {}", connections.len() + 1);
return Ok(Some(new_conn));
},
Err(e) => {
let mut stats = self.stats.lock().map_err(|e| anyhow!("获取统计锁失败: {}", e))?;
stats.connection_errors += 1;
return Err(e);
}
}
}
Ok(None)
}
/// 归还连接到池中
fn return_connection(&self, connection: Connection) {
let mut connections = self.connections.lock().unwrap();
connections.push_back(PooledConnection::new(connection));
let mut stats = self.stats.lock().unwrap();
stats.total_returned += 1;
stats.active_connections = stats.active_connections.saturating_sub(1);
stats.idle_connections += 1;
}
/// 获取连接池统计信息
pub fn get_stats(&self) -> PoolStats {
let stats = self.stats.lock().unwrap();
PoolStats {
total_connections: {
let connections = self.connections.lock().unwrap();
connections.len()
},
active_connections: stats.active_connections,
idle_connections: stats.idle_connections,
total_acquired: stats.total_acquired,
total_returned: stats.total_returned,
acquire_timeouts: stats.acquire_timeouts,
connection_errors: stats.connection_errors,
}
}
/// 关闭连接池
pub fn close(&self) -> Result<()> {
let mut connections = self.connections.lock().map_err(|e| anyhow!("获取连接池锁失败: {}", e))?;
connections.clear();
println!("连接池已关闭");
Ok(())
}
}

View File

@@ -2,11 +2,55 @@ use rusqlite::Connection;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use anyhow::{Result, anyhow};
use std::ops::{Deref, DerefMut};
use crate::infrastructure::connection_pool::{ConnectionPool, ConnectionPoolConfig, PooledConnectionHandle};
/// 统一的数据库连接句柄
/// 可以是单连接模式的 MutexGuard 或连接池模式的 PooledConnectionHandle
pub enum ConnectionHandle<'a> {
/// 单连接模式(兼容模式)
Single(std::sync::MutexGuard<'a, Connection>),
/// 连接池模式(推荐)
Pooled(PooledConnectionHandle),
}
impl<'a> Deref for ConnectionHandle<'a> {
type Target = Connection;
fn deref(&self) -> &Self::Target {
match self {
ConnectionHandle::Single(conn) => conn,
ConnectionHandle::Pooled(conn) => conn.as_ref(),
}
}
}
impl<'a> DerefMut for ConnectionHandle<'a> {
fn deref_mut(&mut self) -> &mut Self::Target {
match self {
ConnectionHandle::Single(conn) => conn,
ConnectionHandle::Pooled(conn) => conn.as_mut(),
}
}
}
/// 数据库管理器
/// 遵循 Tauri 开发规范的数据库设计模式
/// 支持连接池以提高并发性能
pub struct Database {
// 保留原有的单连接模式以兼容现有代码
connection: Arc<Mutex<Connection>>,
// 新增连接池支持
pool: Option<Arc<ConnectionPool>>,
}
/// 数据库连接模式
#[derive(Debug, Clone)]
pub enum ConnectionMode {
/// 单连接模式(兼容模式)
Single,
/// 连接池模式(推荐)
Pool(ConnectionPoolConfig),
}
impl Database {
@@ -23,8 +67,76 @@ impl Database {
Self::new_with_path(db_path.to_str().unwrap())
}
/// 创建带连接池的数据库实例(推荐用于生产环境)
pub fn new_with_pool() -> Result<Self> {
let app_data_dir = dirs::data_dir()
.ok_or_else(|| anyhow!("无法获取应用数据目录"))?
.join("mixvideo");
std::fs::create_dir_all(&app_data_dir)?;
let db_path = app_data_dir.join("mixvideo.db");
// 检查数据库是否被锁定
if let Err(e) = Self::check_database_lock(&db_path) {
eprintln!("数据库锁检查失败: {}", e);
// 尝试清理锁文件
Self::cleanup_database_locks(&db_path)?;
}
Self::new_with_path_and_pool(db_path.to_str().unwrap(), Some(ConnectionPoolConfig::default()))
}
/// 检查数据库锁状态
fn check_database_lock(db_path: &PathBuf) -> Result<()> {
// 检查 WAL 和 SHM 文件
let wal_path = db_path.with_extension("db-wal");
let shm_path = db_path.with_extension("db-shm");
if wal_path.exists() || shm_path.exists() {
println!("检测到 WAL/SHM 文件,数据库可能正在被其他进程使用");
}
// 尝试打开数据库进行快速检查
match Connection::open(db_path) {
Ok(conn) => {
// 尝试执行一个简单的查询
match conn.execute("SELECT 1", []) {
Ok(_) => {
println!("数据库连接检查通过");
Ok(())
},
Err(e) => Err(anyhow!("数据库查询失败: {}", e))
}
},
Err(e) => Err(anyhow!("无法打开数据库: {}", e))
}
}
/// 清理数据库锁文件
fn cleanup_database_locks(db_path: &PathBuf) -> Result<()> {
let wal_path = db_path.with_extension("db-wal");
let shm_path = db_path.with_extension("db-shm");
if wal_path.exists() {
println!("清理 WAL 文件: {:?}", wal_path);
std::fs::remove_file(&wal_path).ok(); // 忽略错误
}
if shm_path.exists() {
println!("清理 SHM 文件: {:?}", shm_path);
std::fs::remove_file(&shm_path).ok(); // 忽略错误
}
Ok(())
}
/// 使用指定路径创建数据库实例(主要用于测试)
pub fn new_with_path(db_path: &str) -> Result<Self> {
Self::new_with_path_and_pool(db_path, None)
}
/// 使用指定路径和连接池配置创建数据库实例
pub fn new_with_path_and_pool(db_path: &str, pool_config: Option<ConnectionPoolConfig>) -> Result<Self> {
let db_path = std::path::PathBuf::from(db_path);
// 打印数据库路径用于调试
@@ -51,14 +163,29 @@ impl Database {
// 配置数据库设置
connection.pragma_update(None, "foreign_keys", "ON")?;
connection.pragma_update(None, "journal_mode", "DELETE")?; // 使用 DELETE 模式而不是 WAL
connection.pragma_update(None, "synchronous", "FULL")?; // 确保数据立即写入磁盘
connection.pragma_update(None, "journal_mode", "WAL")?; // 使用 WAL 模式以支持并发访问
connection.pragma_update(None, "synchronous", "NORMAL")?; // 使用 NORMAL 以提高性能
connection.pragma_update(None, "cache_size", "10000")?; // 增加缓存大小
connection.pragma_update(None, "busy_timeout", "5000")?; // 设置忙等待超时为5秒
println!("Database pragmas configured");
// 创建连接池(如果配置了)
let pool = if let Some(config) = pool_config {
println!("Initializing connection pool with min={}, max={} connections",
config.min_connections, config.max_connections);
let pool = ConnectionPool::new(db_path.to_string_lossy().to_string(), config)?;
println!("Connection pool initialized successfully");
Some(pool)
} else {
println!("Using single connection mode (no pool)");
None
};
let database = Database {
connection: Arc::new(Mutex::new(connection)),
pool,
};
// 初始化数据库表
@@ -93,6 +220,51 @@ impl Database {
}
}
/// 检查是否启用了连接池
pub fn has_pool(&self) -> bool {
self.pool.is_some()
}
/// 从连接池获取连接(推荐)
/// 如果连接池未启用,返回错误
pub fn acquire_from_pool(&self) -> Result<PooledConnectionHandle> {
match &self.pool {
Some(pool) => pool.acquire().map_err(|e| anyhow!("获取连接池连接失败: {}", e)),
None => Err(anyhow!("连接池未启用")),
}
}
/// 尝试从连接池获取连接(非阻塞)
/// 如果连接池未启用或无可用连接,返回 None
pub fn try_acquire_from_pool(&self) -> Result<Option<PooledConnectionHandle>> {
match &self.pool {
Some(pool) => pool.try_acquire().map_err(|e| anyhow!("尝试获取连接池连接失败: {}", e)),
None => Ok(None),
}
}
/// 获取连接(自动选择最佳方式)
/// 如果连接池可用,优先使用连接池
/// 否则使用单连接模式
pub fn get_best_connection(&self) -> Result<ConnectionHandle> {
if let Some(pool) = &self.pool {
// 优先使用连接池
match pool.try_acquire()? {
Some(conn) => return Ok(ConnectionHandle::Pooled(conn)),
None => {
// 连接池中没有可用连接,尝试获取主连接
println!("连接池中没有可用连接,尝试获取主连接");
}
}
}
// 回退到单连接模式
match self.connection.try_lock() {
Ok(conn) => Ok(ConnectionHandle::Single(conn)),
Err(_) => Err(anyhow!("所有数据库连接都被占用")),
}
}
/// 执行数据库操作的辅助方法,自动处理锁的获取和释放
/// 这是推荐的数据库访问方式,可以避免锁竞争问题
pub fn with_connection<T, F>(&self, operation: F) -> Result<T, rusqlite::Error>

View File

@@ -1,6 +1,7 @@
/// 基础设施层模块
/// 遵循 Tauri 开发规范的分层架构设计
pub mod database;
pub mod connection_pool;
pub mod file_system;
pub mod performance;
pub mod event_bus;

View File

@@ -49,6 +49,8 @@ pub fn run() {
commands::database_commands::initialize_database,
commands::database_commands::check_database_connection,
commands::database_commands::force_release_database_connection,
commands::database_commands::get_connection_pool_stats,
commands::database_commands::test_connection_pool_init,
commands::material_commands::import_materials,
commands::material_commands::import_materials_async,
commands::material_commands::select_material_folders,

View File

@@ -23,10 +23,50 @@ pub fn check_database_connection(state: State<AppState>) -> Result<String, Strin
#[tauri::command]
pub fn force_release_database_connection(state: State<AppState>) -> Result<String, String> {
println!("警告:强制释放数据库连接");
// 重新初始化数据库连接
match state.initialize_database() {
Ok(_) => Ok("数据库连接已重新初始化".to_string()),
Err(e) => Err(format!("重新初始化数据库失败: {}", e))
}
}
/// 获取连接池统计信息
#[tauri::command]
pub fn get_connection_pool_stats(state: State<AppState>) -> Result<String, String> {
let database_guard = state.database.lock().map_err(|e| format!("获取数据库失败: {}", e))?;
let database = database_guard.as_ref().ok_or("数据库未初始化")?;
if !database.has_pool() {
return Ok("数据库未启用连接池".to_string());
}
match database.try_acquire_from_pool() {
Ok(Some(_)) => Ok("连接池状态:有可用连接".to_string()),
Ok(None) => Ok("连接池状态:无可用连接".to_string()),
Err(e) => Err(format!("获取连接池状态失败: {}", e)),
}
}
/// 测试连接池初始化
#[tauri::command]
pub fn test_connection_pool_init(_state: State<AppState>) -> Result<String, String> {
use crate::infrastructure::database::Database;
println!("开始测试连接池初始化...");
match Database::new_with_pool() {
Ok(db) => {
println!("连接池初始化成功");
if db.has_pool() {
Ok("连接池测试成功:连接池已启用".to_string())
} else {
Ok("连接池测试警告:连接池未启用".to_string())
}
},
Err(e) => {
eprintln!("连接池初始化失败: {}", e);
Err(format!("连接池测试失败: {}", e))
}
}
}