/** * 多数据源配置文件 * * 架构说明: * ┌─────────────────────────────────────────────────────────┐ * │ 应用服务 │ * ├─────────────────────┬───────────────────────────────────┤ * │ 本地数据库 │ 云数据库 │ * │ (localhost) │ (rm-xxx.rds.aliyuncs.com) │ * │ │ │ * │ ┌──────────────┐ │ ┌────────────────────────────┐ │ * │ │ 用户表 │ │ │ Orders (订单表) │ │ * │ │ 部门表 │ │ │ Products (产品表) │ │ * │ │ 权限表 │ │ │ ... 其他业务表 │ │ * │ │ 配置表 │ │ │ │ │ * │ │ 预警记录 │ │ │ (千万级数据) │ │ * │ └──────────────┘ │ └────────────────────────────┘ │ * │ │ │ * │ 用途: │ 用途: │ * - 用户认证 │ - Cube.js 数据聚合 │ * - 权限管理 │ - 报表查询 │ * - 系统配置 │ - 预警检查 │ * - 预警规则/记录 │ │ * └─────────────────────┴───────────────────────────────────┘ */ const mysql = require('mysql2/promise'); const { getSystemDbRuntimeConfig } = require('../services/systemDbConfig'); const TRANSIENT_MYSQL_ERROR_CODES = new Set([ 'ETIMEDOUT', 'ECONNRESET', 'ECONNREFUSED', 'EPIPE', 'PROTOCOL_CONNECTION_LOST', 'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR', 'PROTOCOL_ENQUEUE_AFTER_QUIT', 'PROTOCOL_PACKETS_OUT_OF_ORDER', ]); function parsePositiveInt(value, fallback) { const parsed = Number.parseInt(value, 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; } function isTransientMysqlError(error) { if (!error) { return false; } const message = String(error.message || '').toLowerCase(); const code = String(error.code || ''); return ( TRANSIENT_MYSQL_ERROR_CODES.has(code) || message.includes('timed out') || message.includes('read etimedout') || message.includes('connect etimedout') || message.includes('connection lost') || message.includes('pool is closed') || message.includes('closed state') || message.includes('socket hang up') ); } const LOCAL_DB_CONNECT_TIMEOUT_MS = parsePositiveInt(process.env.LOCAL_DB_CONNECT_TIMEOUT_MS, 5000); const LOCAL_DB_CONNECTION_LIMIT = parsePositiveInt(process.env.LOCAL_DB_CONNECTION_LIMIT, 6); const LOCAL_DB_QUEUE_LIMIT = parsePositiveInt(process.env.LOCAL_DB_QUEUE_LIMIT, 50); const LOCAL_DB_MAX_IDLE = parsePositiveInt(process.env.LOCAL_DB_MAX_IDLE, LOCAL_DB_CONNECTION_LIMIT); const LOCAL_DB_IDLE_TIMEOUT_MS = parsePositiveInt(process.env.LOCAL_DB_IDLE_TIMEOUT_MS, 60000); const LOCAL_DB_QUERY_RETRY_COUNT = parsePositiveInt(process.env.LOCAL_DB_QUERY_RETRY_COUNT, 1); const LOCAL_DB_QUERY_TIMEOUT_MS = parsePositiveInt(process.env.LOCAL_DB_QUERY_TIMEOUT_MS, 60000); function withLocalQueryTimeout(sql) { if (!LOCAL_DB_QUERY_TIMEOUT_MS) { return sql; } return { sql, timeout: LOCAL_DB_QUERY_TIMEOUT_MS, }; } function createCloudPool() { return mysql.createPool({ host: process.env.CLOUD_DB_HOST, port: parseInt(process.env.CLOUD_DB_PORT || '3306'), user: process.env.CLOUD_DB_USER, password: process.env.CLOUD_DB_PASSWORD, database: process.env.CLOUD_DB_NAME, charset: 'utf8mb4', waitForConnections: true, connectionLimit: 10, queueLimit: 0, ssl: process.env.CLOUD_DB_SSL === 'true' ? { rejectUnauthorized: true } : false, }); } function createLocalPool() { const config = getSystemDbRuntimeConfig(); return mysql.createPool({ host: config.host, port: parseInt(config.port || '3306', 10), user: config.username, password: config.password, database: config.database, charset: 'utf8mb4', waitForConnections: true, connectionLimit: LOCAL_DB_CONNECTION_LIMIT, maxIdle: LOCAL_DB_MAX_IDLE, idleTimeout: LOCAL_DB_IDLE_TIMEOUT_MS, queueLimit: LOCAL_DB_QUEUE_LIMIT, connectTimeout: LOCAL_DB_CONNECT_TIMEOUT_MS, enableKeepAlive: true, keepAliveInitialDelay: 0, }); } let cloudPool = createCloudPool(); let localPool = createLocalPool(); async function resetPool(source) { const isLocal = source === 'local'; const activePool = isLocal ? localPool : cloudPool; if (activePool) { try { await activePool.end(); } catch (error) { console.warn(`[database:${source}] 重置连接池时关闭失败: ${error.message}`); } } if (isLocal) { localPool = createLocalPool(); dataSource.local.pool = localPool; } else { cloudPool = createCloudPool(); dataSource.cloud.pool = cloudPool; } } async function runQueryWithRetry(source, runner, options = {}) { const maxAttempts = options.maxAttempts || 1; for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { try { const activePool = source === 'local' ? localPool : cloudPool; return await runner(activePool); } catch (error) { const shouldRetry = source === 'local' && attempt < maxAttempts && isTransientMysqlError(error); if (!shouldRetry) { throw error; } console.warn(`[database:${source}] 查询失败,重置连接池后重试 (${attempt}/${maxAttempts}): ${error.message}`); await resetPool(source); } } return []; } // ============================================ // 数据源管理器 // ============================================ const dataSource = { // 云数据库 (业务数据 - 只读为主) cloud: { name: 'cloud', pool: cloudPool, isReadOnly: true, // 测试连接 async test() { try { const connection = await cloudPool.getConnection(); await connection.ping(); connection.release(); return { success: true, message: '业务数据库连接成功' }; } catch (error) { return { success: false, message: `连接失败: ${error.message}` }; } }, // 执行查询 async query(sql, params = []) { const [rows] = await runQueryWithRetry('cloud', (pool) => pool.execute(sql, params)); return rows; }, }, // 系统数据库 (系统配置 + 用户数据 + 汇总结果 - 读写) local: { name: 'local', pool: localPool, isReadOnly: false, // 测试连接 async test() { try { const connection = await runQueryWithRetry( 'local', (pool) => pool.getConnection(), { maxAttempts: LOCAL_DB_QUERY_RETRY_COUNT } ); await connection.ping(); connection.release(); return { success: true, message: '系统数据库连接成功' }; } catch (error) { return { success: false, message: `连接失败: ${error.message}` }; } }, // 执行查询 async query(sql, params = []) { // 使用 query 而不是 execute,因为 execute 对 LIMIT 参数有限制 const [rows] = await runQueryWithRetry( 'local', (pool) => pool.query(withLocalQueryTimeout(sql), params), { maxAttempts: LOCAL_DB_QUERY_RETRY_COUNT } ); return rows; }, // 执行插入 - 支持两种方式: // 1. insert(sql, params) - 原始 SQL // 2. insert(tableName, data) - 表名 + 数据对象 async insert(tableOrSql, dataOrParams = []) { let sql, params; if (typeof dataOrParams === 'object' && !Array.isArray(dataOrParams)) { // 表名 + 数据对象方式 const tableName = tableOrSql; const data = dataOrParams; const columns = Object.keys(data); const values = Object.values(data); const placeholders = columns.map(() => '?').join(', '); // 对保留字使用反引号 const quotedColumns = columns.map(c => `\`${c}\``).join(', '); sql = `INSERT INTO ${tableName} (${quotedColumns}) VALUES (${placeholders})`; params = values; } else { // 原始 SQL 方式 sql = tableOrSql; params = dataOrParams; } const [result] = await runQueryWithRetry( 'local', (pool) => pool.query(withLocalQueryTimeout(sql), params), { maxAttempts: LOCAL_DB_QUERY_RETRY_COUNT } ); return { insertId: result.insertId }; }, // 执行更新 - 支持两种方式: // 1. update(sql, params) - 原始 SQL // 2. update(tableName, id, data) - 表名 + ID + 数据对象 async update(tableOrSql, idOrParams, data) { let sql, params; if (data !== undefined && typeof data === 'object') { // 表名 + ID + 数据对象方式 const tableName = tableOrSql; const id = idOrParams; const columns = Object.keys(data); const values = Object.values(data); // 对保留字使用反引号 const setClause = columns.map(c => `\`${c}\` = ?`).join(', '); sql = `UPDATE ${tableName} SET ${setClause} WHERE id = ?`; params = [...values, id]; } else { // 原始 SQL 方式 sql = tableOrSql; params = idOrParams || []; } const [result] = await runQueryWithRetry( 'local', (pool) => pool.query(withLocalQueryTimeout(sql), params), { maxAttempts: LOCAL_DB_QUERY_RETRY_COUNT } ); return result.affectedRows; }, }, // 获取指定数据源 get(source) { return source === 'cloud' ? this.cloud : this.local; }, // 关闭所有连接 async close() { await cloudPool.end(); await localPool.end(); }, }; module.exports = dataSource;