AI生成的数据库监控系统，竟比手工开发的还要智能？

俊才

发布于 2026-01-27 14:52:54

1260

"数据库又慢了！"这已经成为我们DBA团队的日常噩梦。随着业务快速发展，公司同时使用Oracle和MySQL数据库，性能问题频发。每次业务卡顿，我们都得像救火队员一样，手动登录各个数据库排查，效率低下且被动。今天我用AI帮忙生成了一个统一的数据库慢SQL监控系统，不但开发效率高而且还提供了比较智能化思路。

感兴趣的同学可以联系我获取源码，一起共同改造提升、丰富功能。

一、背景

1. 痛点：分散监控的困境

之前我已经写过有两套独立的监控方案：

Oracle监控：通过定期查询vsession和vsql视图，找出运行时间超过阈值的SQL语句
MySQL监控：使用information_schema.PROCESSLIST命令，筛选长时间运行的SQL语句

虽然也都配置了监控到就发送预警信息，但是这种分散的监控方式还是存在明显问题：

排查效率低：需要在不同工具间切换，无法统一查看
历史数据缺失：难以进行趋势分析和问题追溯

2. 转折点：统一监控及可视化的想法

直到最近，我在做日常巡检及统计的时候，我下定决心：必须建立一个统一的数据库监控平台给自己用，来提升效率及可视化展示，方便分析及处理。

二、设计过程

1. 第一版设计思路

最开始我的设想比较简单，就是通过采集程序将各个数据采集、存储在数据库的表里

2. 第二步优化

以MySQL为例，在原先的MySQL的监控程序基础上添加前端展示页面

3. 最终前端优化

最终采集了几个实例的数据库后发现数据量还是比较多的，整个页面就显得很长，因此考虑进行分页展示SQL明细。

三、程序设计

根据上面的思路，最终在AI的帮助下，完成了总体的程序设计，基本满足了SQL展示及统计的需求，后面还有需求会在上面进行迭代。

1. 项目结构

由于涉及监控采集脚本、Flask主程序、配置文件（可选，我最终去掉了这文件，简化一下）、静态资源（前端页面、样式文件）、日志目录（运行日志），最终的项目结构如下：

long_sql_monitor/
├── db_long_sql_monitor.py # 监控采集程序
├── app.py                 # Flask主程序
├── requirements.txt       # 依赖文件
├── config.py             # 配置文件
├── static/
│   ├── index.html        # 前端页面
│   └── css/              # 样式文件
└── logs/                 # 日志目录

2. 核心技术选型

后端：Python + Flask + 数据库连接池
前端：原生HTML/CSS/JS + ECharts图表
数据库：MySQL（存储监控数据）+ 多类型业务数据库
调度：Schedule轻量级定时任务
消息：飞书机器人 + 邮件 + SMS

3. 核心模块代码

3.1 管理模块

主要涉及数据库实例信息的获取，动态条件构建，灵活支持多种查询场景；另外考虑用缓存机制减少数据库访问压力，且统一的配置管理入口

class ConfigManager:
    def get_database_instances(self, db_type=None):
        """智能获取数据库实例配置，支持缓存优化"""
        instances = []
        conn = self._get_mysql_connection()
        try:
            with conn.cursor() as cursor:
                # 动态构建查询条件
                conditions = ["db_ip IS NOT NULL", "is_active = 1"]
                params = []
                if db_type and db_type in ['Oracle', 'MySQL']:
                    conditions.append("db_type = %s")
                    params.append(db_type)
                where_clause = " AND ".join(conditions)
                sql = f"""
                SELECT id, db_project, db_ip, db_port, instance_name,
                       db_user, db_password, db_type, monitor_interval
                FROM db_instance_info 
                WHERE {where_clause}
                ORDER BY db_type, db_project
                """
                cursor.execute(sql, params)
                return cursor.fetchall()
        except Exception as e:
            logging.error(f"配置获取失败: {str(e)}")
            return []

3.2 多数据库适配引擎

因为后续还会加其他的数据库，因此设计统一接口，新增数据库类型只需实现对应方法。连接池管理，避免频繁创建销毁连接；另外做了结果标准化，便于后续处理，对于的代码片段如下：

class DatabaseAdapter:
    def get_long_running_sql(self, instance_info):
        """统一接口获取不同数据库的长时SQL"""
        db_type = instance_info['db_type'].lower()
        if db_type == 'oracle':
            return self._get_oracle_long_sql(instance_info)
        elif db_type == 'mysql':
            return self._get_mysql_long_sql(instance_info)
        else:
            raise ValueError(f"不支持的数据库类型: {db_type}")
    def _get_oracle_long_sql(self, instance_info):
        """Oracle长SQL检测实现"""
        conn = self._connect_oracle(instance_info)
        cursor = conn.cursor()
        sql = """
        SELECT s.sid, s.serial#, s.username, s.sql_id,
               sq.sql_text, (SYSDATE - s.sql_exec_start) * 1440 as elapsed_minutes
        FROM v$session s
        JOIN v$sql sq ON s.sql_id = sq.sql_id
        WHERE s.status = 'ACTIVE' 
          AND (SYSDATE - s.sql_exec_start) * 1440 > 1
          AND s.username IS NOT NULL
        """
        cursor.execute(sql)
        return self._format_oracle_results(cursor.fetchall())
    def _get_mysql_long_sql(self, instance_info):
        """MySQL长SQL检测实现"""
        conn = self._connect_mysql(instance_info)
        cursor = conn.cursor()
        sql = """
        SELECT id, user, host, db, command, time, state, info
        FROM information_schema.processlist 
        WHERE command != 'Sleep' 
          AND time > 1
          AND info IS NOT NULL
        """
        cursor.execute(sql)
        return self._format_mysql_results(cursor.fetchall())

3.3 智能预警引擎

做了几个级别，进行多维度评分，作为评估问题严重性的参考（可按照自己的业务情况调整）。由于有几个库慢SQL比较多，因此加了防骚扰机制，避免告警风暴。

class SmartAlertEngine:
    def analyze_sql_severity(self, sql_info):
        """智能分析SQL严重程度"""
        base_score = 0
        # 运行时长评分（指数级增长）
        if sql_info['elapsed_minutes'] > 60:    # 1小时以上
            base_score += 100
        elif sql_info['elapsed_minutes'] > 30:  # 30分钟以上
            base_score += 50
        elif sql_info['elapsed_minutes'] > 10:  # 10分钟以上
            base_score += 20
        elif sql_info['elapsed_minutes'] > 5:   # 5分钟以上
            base_score += 10
        else:                                   # 1-5分钟
            base_score += 5
        # 资源消耗评分
        if hasattr(sql_info, 'cpu_time') and sql_info['cpu_time'] > 3600:
            base_score += 30
        if hasattr(sql_info, 'physical_reads') and sql_info['physical_reads'] > 10000:
            base_score += 25
        # 业务重要性加权
        if '核心业务' in sql_info.get('db_project', ''):
            base_score *= 1.5
        return self._score_to_severity(base_score)
    def should_alert(self, instance_ip, severity, alert_history):
        """智能防骚扰判断"""
        from datetime import datetime, timedelta
        # 同一实例相同严重程度的告警，5分钟内不重复
        key = f"{instance_ip}_{severity}"
        last_alert = alert_history.get(key)
        if last_alert and datetime.now() - last_alert < timedelta(minutes=5):
            return False
        # 更新告警历史
        alert_history[key] = datetime.now()
        return True

3.4 实时数据存储模块

在数据存储这块，目前考虑的安装如下方式提升存储速度和异常处理：

批量插入操作，提升写入性能
事务分批提交，避免长事务锁定
数据截断处理，防止超长字段报错

class DataManager:
    def batch_save_sql_records(self, instance_id, sql_records):
        """批量保存SQL监控记录，高性能设计"""
        if not sql_records:
            return
        conn = self._get_connection()
        try:
            with conn.cursor() as cursor:
                # 批量插入数据
                sql = """
                INSERT INTO long_running_sql_log 
                (db_instance_id, session_id, username, sql_text, 
                 elapsed_minutes, detect_time, db_type)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                """
                # 分批处理，避免单次事务过大
                batch_size = 100
                for i in range(0, len(sql_records), batch_size):
                    batch = sql_records[i:i + batch_size]
                    values = [
                        (instance_id, rec['session_id'], rec['username'],
                         rec['sql_text'][:1000], rec['elapsed_minutes'],
                         datetime.now(), rec.get('db_type', 'Unknown'))
                        for rec in batch
                    ]
                    cursor.executemany(sql, values)
                conn.commit()
                logging.info(f"成功保存 {len(sql_records)} 条监控记录")
        except Exception as e:
            logging.error(f"数据保存失败: {str(e)}")
            conn.rollback()
        finally:
            conn.close()

3.5 RESTful API 接口

API设计这块不专业，就按照如下原则让AI帮忙生成了：

统一的响应格式
完善的错误处理
灵活的参数支持
分页查询优化

@app.route('/api/long_sql', methods=['GET'])
def get_long_running_sql():
    """获取长时SQL数据接口"""
    try:
        # 参数解析和验证
        hours = request.args.get('hours', 24, type=int)
        db_type = request.args.get('db_type', type=str)
        page = request.args.get('page', 1, type=int)
        page_size = min(request.args.get('page_size', 20, type=int), 100)
        # 数据查询
        sql_records = data_manager.query_long_sql(
            hours=hours, 
            db_type=db_type,
            page=page, 
            page_size=page_size
        )
        # 构建响应
        return jsonify({
            'success': True,
            'data': sql_records,
            'pagination': {
                'page': page,
                'page_size': page_size,
                'total': len(sql_records)
            },
            'timestamp': datetime.now().isoformat()
        })
    except Exception as e:
        logging.error(f"API查询失败: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

3.6 实时前端展示

前端按照如下特性生成的代码：

实时数据更新，无需手动刷新
智能页面检测，隐藏时暂停更新
响应式设计，支持多终端访问
丰富的可视化图表

class RealTimeDashboard {
    constructor() {
        this.charts = {};
        this.autoRefresh = true;
    }
    // 初始化图表
    initCharts() {
        // 实例分布图表
        this.charts.instanceChart = echarts.init(document.getElementById('instanceChart'));
        this.charts.instanceChart.setOption({
            title: { text: '实例监控统计' },
            tooltip: { trigger: 'axis' },
            xAxis: { type: 'category', data: [] },
            yAxis: { type: 'value' },
            series: [{ data: [], type: 'bar' }]
        });
        // 趋势分析图表
        this.charts.trendChart = echarts.init(document.getElementById('trendChart'));
    }
    // 实时数据更新
    async updateDashboard() {
        try {
            const [stats, sqlData] = await Promise.all([
                this.fetchStatistics(),
                this.fetchSQLData()
            ]);
            this.updateStatsCards(stats);
            this.updateInstanceChart(stats.instanceStats);
            this.updateTrendChart(stats.trendData);
            this.updateSQLTable(sqlData);
        } catch (error) {
            this.showError('数据更新失败: ' + error.message);
        }
    }
    // 智能自动刷新
    startAutoRefresh() {
        setInterval(() => {
            if (this.autoRefresh && this.isPageVisible()) {
                this.updateDashboard();
            }
        }, 30000); // 30秒刷新
    }
}

4. 运行效果

4.1 启动应用

本次涉及采集程序(db_long_sql_monitor.py)及应用程序（app.py）,因此需要启动这2个程序

nohup  python3 db_long_sql_monitor.py & 
nohup python3 app.py &

4.2 运行的效果如下

当前实现的功能如下：实例个数、慢SQL数量、平均市场、严重告警数量等。也可以进去时间范围（最近1h,24h、最近7天，30天，全部等），实例筛选、最小运行时间（后续在调整更细粒度及范围的），分页展示行数等

统计模块包含按照实例统计及时间统计趋势。最终具体的SQL列表可以在上面进行筛选、每页默认20行记录。

5. 结语

通过这个AI生成的数据库监控系统，我们见证了AI如何重塑传统运维工作流。从需求描述到完整系统，AI不仅大幅提升了开发效率，更带来了意想不到的智能化水平。随着AI代码生成能力的持续进化，我们有理由相信：未来的运维系统将更加智能、自适应甚至自修复。而作为技术人，我们要做的不是抗拒变化，而是主动拥抱这场效率革命，将AI转化为提升运维效能的神兵利器。

你是否也曾尝试用AI生成运维脚本或系统？欢迎在评论区分享你的实践经验和心得体会！如果你对文中的监控系统感兴趣，也可以留言获取更多实现细节。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2026-01-19，如有侵权请联系 cloudcommunity@tencent.com 删除

开发