Python中MySQL监控与日志配置实战指南

# 检查mysql是否安装
mysql --version

# 如果未安装，使用apt安装（其他系统请参考官方文档）
sudo apt update
sudo apt install mysql-server mysql-client

# 启动mysql服务
sudo systemctl start mysql
sudo systemctl enable mysql

# 安全初始化（设置root密码等）
sudo mysql_secure_installation

# 创建虚拟环境（推荐）
python -m venv mysql-monitor-env
source mysql-monitor-env/bin/activate  # linux/mac
# 或 mysql-monitor-env\scripts\activate  # windows

# 安装pymysql
pip install pymysql

# 安装额外的工具库，用于后续的监控数据可视化
pip install matplotlib pandas

# create_test_data.py
import pymysql
import random
from datetime import datetime, timedelta

def create_test_database():
    """创建测试数据库和表，并插入模拟数据"""
    
    # 连接mysql（请替换为你的实际密码）
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='your_password',  # 改成你的mysql root密码
        charset='utf8mb4',
        cursorclass=pymysql.cursors.dictcursor
    )
    
    try:
        with connection.cursor() as cursor:
            # 创建数据库
            cursor.execute("create database if not exists ecommerce_monitor")
            cursor.execute("use ecommerce_monitor")
            
            # 创建用户表
            cursor.execute("""
                create table if not exists users (
                    id int auto_increment primary key,
                    username varchar(50) not null unique,
                    email varchar(100) not null,
                    created_at timestamp default current_timestamp,
                    index idx_username (username),
                    index idx_created_at (created_at)
                )
            """)
            
            # 创建订单表（故意不加索引，用于演示慢查询）
            cursor.execute("""
                create table if not exists orders (
                    id int auto_increment primary key,
                    user_id int not null,
                    amount decimal(10, 2) not null,
                    status enum('pending', 'paid', 'shipped', 'delivered') default 'pending',
                    created_at timestamp default current_timestamp,
                    foreign key (user_id) references users(id)
                    # 注意：这里故意不在user_id和created_at上加索引！
                )
            """)
            
            # 插入测试用户数据
            print("插入用户数据...")
            users = []
            for i in range(1, 1001):  # 1000个用户
                users.append((f'user{i}', f'user{i}@example.com'))
            
            cursor.executemany(
                "insert into users (username, email) values (%s, %s)",
                users
            )
            
            # 插入测试订单数据（更多数据，用于模拟真实场景）
            print("插入订单数据...")
            orders = []
            start_date = datetime.now() - timedelta(days=365)
            
            for i in range(1, 50001):  # 5万条订单
                user_id = random.randint(1, 1000)
                amount = round(random.uniform(10.0, 1000.0), 2)
                days_ago = random.randint(0, 365)
                order_date = start_date + timedelta(days=days_ago)
                
                # 随机状态
                status = random.choice(['pending', 'paid', 'shipped', 'delivered'])
                
                orders.append((
                    user_id,
                    amount,
                    status,
                    order_date.strftime('%y-%m-%d %h:%m:%s')
                ))
            
            # 分批插入，避免单次sql太大
            batch_size = 1000
            for i in range(0, len(orders), batch_size):
                batch = orders[i:i+batch_size]
                cursor.executemany(
                    """insert into orders (user_id, amount, status, created_at) 
                       values (%s, %s, %s, %s)""",
                    batch
                )
                connection.commit()
                print(f"已插入 {min(i+batch_size, len(orders))}/{len(orders)} 条订单")
            
            print("测试数据创建完成！")
            
    finally:
        connection.close()

if __name__ == "__main__":
    create_test_database()

# 编辑mysql配置文件
sudo nano /etc/mysql/mysql.conf.d/mysqld.cnf  # ubuntu路径，其他系统可能不同

# 在[mysqld]部分添加或修改以下配置：
"""
[mysqld]
# 开启慢查询日志
slow_query_log = 1

# 指定慢查询日志文件路径
slow_query_log_file = /var/log/mysql/mysql-slow.log

# 设置慢查询阈值（单位：秒），这里设为1秒，生产环境通常设2-3秒
long_query_time = 1

# 记录未使用索引的查询（即使执行时间没超过阈值）
log_queries_not_using_indexes = 1

# 每分钟最多记录多少条慢查询，避免日志爆炸
log_throttle_queries_not_using_indexes = 10
"""

# 保存后重启mysql
sudo systemctl restart mysql

# generate_slow_queries.py
import pymysql
import time

def generate_slow_queries():
    """执行一些会触发慢查询的sql"""
    
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='your_password',
        database='ecommerce_monitor',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.dictcursor
    )
    
    try:
        with connection.cursor() as cursor:
            print("执行可能较慢的查询...")
            
            # 查询1：全表扫描（orders表没有user_id索引）
            start = time.time()
            cursor.execute("""
                select * from orders 
                where user_id = 500 
                and created_at > '2023-01-01'
                order by created_at desc
            """)
            result1 = cursor.fetchall()
            elapsed1 = time.time() - start
            print(f"查询1（无索引条件查询）耗时: {elapsed1:.3f}秒，返回 {len(result1)} 条记录")
            
            # 查询2：复杂联表查询
            start = time.time()
            cursor.execute("""
                select u.username, count(o.id) as order_count, sum(o.amount) as total_amount
                from users u
                left join orders o on u.id = o.user_id
                where o.created_at > '2023-06-01'
                group by u.id
                having order_count > 5
                order by total_amount desc
                limit 20
            """)
            result2 = cursor.fetchall()
            elapsed2 = time.time() - start
            print(f"查询2（复杂联表分组）耗时: {elapsed2:.3f}秒")
            
            # 查询3：使用索引的快速查询（作为对比）
            start = time.time()
            cursor.execute("select * from users where username = 'user500'")
            result3 = cursor.fetchall()
            elapsed3 = time.time() - start
            print(f"查询3（使用索引查询）耗时: {elapsed3:.3f}秒")
            
    finally:
        connection.close()

if __name__ == "__main__":
    generate_slow_queries()

# 查看慢查询日志（需要sudo权限）
sudo tail -100 /var/log/mysql/mysql-slow.log

# 使用mysqldumpslow工具分析（mysql自带）
sudo mysqldumpslow /var/log/mysql/mysql-slow.log -t 10 -s at

# 输出示例：
"""
count: 3  time=1.23s (3s)  lock=0.00s (0s)  rows=152.7 (458), root[root]@localhost
  select * from orders where user_id = n and created_at > 's' order by created_at desc
"""

# add_indexes.py
import pymysql

def add_necessary_indexes():
    """根据慢查询分析添加缺失的索引"""
    
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='your_password',
        database='ecommerce_monitor',
        charset='utf8mb4'
    )
    
    try:
        with connection.cursor() as cursor:
            print("添加缺失的索引...")
            
            # 为orders表的user_id添加索引
            cursor.execute("""
                alter table orders 
                add index idx_user_id (user_id),
                add index idx_created_at (created_at),
                add index idx_user_created (user_id, created_at)
            """)
            
            print("索引添加完成！")
            
            # 验证优化效果
            print("\n验证优化效果：")
            cursor.execute("explain select * from orders where user_id = 500 and created_at > '2023-01-01'")
            explain_result = cursor.fetchone()
            
            print(f"查询执行计划：")
            print(f"- 使用的索引: {explain_result.get('key', '无')}")
            print(f"- 扫描行数: {explain_result.get('rows', '未知')}")
            print(f"- 查询类型: {explain_result.get('type', '未知')}")
            
    finally:
        connection.close()

if __name__ == "__main__":
    add_necessary_indexes()

# performance_monitor.py
import pymysql
import time
import pandas as pd
from datetime import datetime

class mysqlperformancemonitor:
    """mysql性能监控器"""
    
    def __init__(self, host='localhost', user='root', password='', database=''):
        self.connection = pymysql.connect(
            host=host,
            user=user,
            password=password,
            database=database,
            charset='utf8mb4',
            cursorclass=pymysql.cursors.dictcursor
        )
    
    def get_slow_queries_summary(self):
        """获取慢查询摘要统计"""
        with self.connection.cursor() as cursor:
            cursor.execute("""
                select 
                    digest_text as query_pattern,
                    count_star as exec_count,
                    avg_timer_wait/1000000000000 as avg_time_sec,
                    max_timer_wait/1000000000000 as max_time_sec,
                    sum_rows_examined as rows_examined_total,
                    sum_rows_sent as rows_sent_total
                from performance_schema.events_statements_summary_by_digest
                where digest_text is not null
                and avg_timer_wait > 1000000000  # 大于1毫秒
                order by avg_time_sec desc
                limit 10
            """)
            return cursor.fetchall()
    
    def get_connection_stats(self):
        """获取连接统计"""
        with self.connection.cursor() as cursor:
            cursor.execute("""
                select 
                    user as user,
                    host as host,
                    count(*) as connection_count,
                    group_concat(command) as commands
                from information_schema.processlist
                where command != 'sleep'
                group by user, host
            """)
            return cursor.fetchall()
    
    def get_table_access_stats(self, hours=24):
        """获取表访问统计"""
        with self.connection.cursor() as cursor:
            # 注意：这个查询需要开启某些consumer，默认可能没有数据
            cursor.execute("""
                select 
                    object_schema as db_name,
                    object_name as table_name,
                    count_read as read_count,
                    count_write as write_count,
                    count_fetch as fetch_count
                from performance_schema.table_io_waits_summary_by_table
                where count_star > 0
                order by count_star desc
                limit 10
            """)
            return cursor.fetchall()
    
    def monitor_loop(self, interval=60, duration=300):
        """监控循环，定期收集性能数据"""
        print(f"开始性能监控，每{interval}秒采样一次，持续{duration}秒...")
        
        data_points = []
        start_time = time.time()
        
        while time.time() - start_time < duration:
            timestamp = datetime.now()
            
            # 收集各种性能指标
            slow_queries = self.get_slow_queries_summary()
            connections = self.get_connection_stats()
            
            # 记录数据点
            data_point = {
                'timestamp': timestamp,
                'slow_query_count': len(slow_queries),
                'active_connections': sum(c['connection_count'] for c in connections),
                'top_slow_query_time': slow_queries[0]['avg_time_sec'] if slow_queries else 0
            }
            
            data_points.append(data_point)
            print(f"[{timestamp}] 慢查询数: {data_point['slow_query_count']}, "
                  f"活跃连接: {data_point['active_connections']}")
            
            time.sleep(interval)
        
        # 转换为dataframe便于分析
        df = pd.dataframe(data_points)
        return df
    
    def close(self):
        self.connection.close()

# 使用示例
if __name__ == "__main__":
    monitor = mysqlperformancemonitor(
        host='localhost',
        user='root',
        password='your_password',
        database='ecommerce_monitor'
    )
    
    try:
        # 获取一次性的性能快照
        print("=== 当前慢查询top 10 ===")
        slow_queries = monitor.get_slow_queries_summary()
        for i, query in enumerate(slow_queries[:5], 1):
            print(f"{i}. {query['query_pattern'][:80]}...")
            print(f"   平均耗时: {query['avg_time_sec']:.3f}s, 执行次数: {query['exec_count']}")
        
        print("\n=== 当前连接统计 ===")
        connections = monitor.get_connection_stats()
        for conn in connections:
            print(f"用户: {conn['user']}, 连接数: {conn['connection_count']}")
        
        # 运行监控循环（生产环境可以改为后台任务）
        # df = monitor.monitor_loop(interval=10, duration=60)
        # print(f"\n监控数据摘要:\n{df.describe()}")
        
    finally:
        monitor.close()

# db_connection_pool.py
import pymysql
import logging
import time
from threading import lock
from contextlib import contextmanager
from datetime import datetime, timedelta

# 配置日志
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.filehandler('mysql_operations.log'),
        logging.streamhandler()
    ]
)
logger = logging.getlogger('mysqlmonitor')

class healthyconnectionpool:
    """带健康检查的mysql连接池"""
    
    def __init__(self, max_connections=10, **kwargs):
        self.max_connections = max_connections
        self.connection_args = kwargs
        self.pool = []
        self.in_use = set()
        self.lock = lock()
        self.last_health_check = datetime.min
        
        # 初始化连接池
        self._initialize_pool()
        
        logger.info(f"连接池初始化完成，最大连接数: {max_connections}")
    
    def _initialize_pool(self):
        """初始化连接池"""
        for _ in range(min(3, self.max_connections)):
            conn = self._create_connection()
            if conn:
                self.pool.append(conn)
    
    def _create_connection(self):
        """创建新连接"""
        try:
            conn = pymysql.connect(**self.connection_args)
            
            # 设置连接属性，便于追踪
            with conn.cursor() as cursor:
                cursor.execute("set @python_client_id = %s", 
                             (f"pool_conn_{len(self.pool)}",))
            
            logger.debug(f"创建新数据库连接: {conn.server_version}")
            return conn
        except exception as e:
            logger.error(f"创建数据库连接失败: {e}")
            return none
    
    def _health_check(self):
        """定期健康检查"""
        now = datetime.now()
        if now - self.last_health_check < timedelta(minutes=5):
            return
        
        with self.lock:
            healthy_connections = []
            for conn in self.pool:
                try:
                    with conn.cursor() as cursor:
                        cursor.execute("select 1")
                        cursor.fetchone()
                    healthy_connections.append(conn)
                except exception as e:
                    logger.warning(f"连接健康检查失败，关闭异常连接: {e}")
                    try:
                        conn.close()
                    except:
                        pass
            
            # 补充连接
            while len(healthy_connections) < self.max_connections:
                new_conn = self._create_connection()
                if new_conn:
                    healthy_connections.append(new_conn)
                else:
                    break
            
            self.pool = healthy_connections
            self.last_health_check = now
            
            logger.info(f"健康检查完成，活跃连接数: {len(self.pool)}")
    
    @contextmanager
    def get_connection(self):
        """获取连接（上下文管理器方式）"""
        self._health_check()
        
        conn = none
        start_time = time.time()
        
        with self.lock:
            if self.pool:
                conn = self.pool.pop()
            elif len(self.in_use) < self.max_connections:
                conn = self._create_connection()
            
            if conn:
                self.in_use.add(id(conn))
        
        if not conn:
            wait_time = time.time() - start_time
            logger.error(f"获取数据库连接超时，等待{wait_time:.2f}秒后仍无可用连接")
            raise exception("数据库连接池耗尽")
        
        try:
            # 记录连接获取
            logger.debug(f"获取数据库连接，当前使用中: {len(self.in_use)}")
            
            # 执行查询前的准备
            with conn.cursor() as cursor:
                cursor.execute("set @query_start_time = now(6)")
            
            yield conn
            
        except pymysql.error as e:
            # 记录数据库错误
            error_code, error_msg = e.args
            logger.error(f"数据库操作错误 [{error_code}]: {error_msg}")
            
            # 根据错误类型决定是否关闭连接
            if error_code in (2006, 2013):  # 连接相关错误
                logger.warning("连接异常，将关闭并创建新连接")
                try:
                    conn.close()
                except:
                    pass
                conn = self._create_connection()
            raise
            
        finally:
            # 记录查询执行时间
            try:
                with conn.cursor() as cursor:
                    cursor.execute("select timestampdiff(microsecond, @query_start_time, now(6)) / 1000000 as exec_time")
                    result = cursor.fetchone()
                    exec_time = result['exec_time'] if result else 0
                    
                    if exec_time > 1.0:  # 超过1秒的记录为慢查询
                        logger.warning(f"慢查询检测: 执行时间 {exec_time:.3f}秒")
            except:
                exec_time = 0
            
            # 归还连接
            with self.lock:
                if conn and conn.open:
                    self.pool.append(conn)
                if id(conn) in self.in_use:
                    self.in_use.remove(id(conn))
            
            logger.debug(f"归还数据库连接，执行时间: {exec_time:.3f}秒")
    
    def close_all(self):
        """关闭所有连接"""
        with self.lock:
            for conn in self.pool:
                try:
                    conn.close()
                except:
                    pass
            self.pool.clear()
            self.in_use.clear()
        
        logger.info("连接池已关闭所有连接")

# 使用示例
def example_usage():
    """使用带监控的连接池示例"""
    
    # 创建连接池
    pool = healthyconnectionpool(
        max_connections=5,
        host='localhost',
        user='root',
        password='your_password',
        database='ecommerce_monitor',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.dictcursor
    )
    
    try:
        # 示例1：正常查询
        with pool.get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute("select count(*) as count from orders")
                result = cursor.fetchone()
                print(f"订单总数: {result['count']}")
        
        # 示例2：事务操作
        with pool.get_connection() as conn:
            try:
                with conn.cursor() as cursor:
                    # 开始事务
                    cursor.execute("start transaction")
                    
                    # 插入新订单
                    cursor.execute(
                        "insert into orders (user_id, amount, status) values (%s, %s, %s)",
                        (1, 99.99, 'pending')
                    )
                    
                    # 更新用户统计（模拟业务逻辑）
                    cursor.execute(
                        "update users set email = %s where id = %s",
                        ('updated@example.com', 1)
                    )
                    
                    # 提交事务
                    conn.commit()
                    logger.info("事务提交成功")
                    
            except exception as e:
                conn.rollback()
                logger.error(f"事务回滚: {e}")
                raise
        
        # 示例3：批量查询（模拟业务高峰）
        import concurrent.futures
        
        def query_user_orders(user_id):
            """查询用户订单"""
            with pool.get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(
                        "select * from orders where user_id = %s limit 10",
                        (user_id,)
                    )
                    return cursor.fetchall()
        
        # 模拟并发查询
        with concurrent.futures.threadpoolexecutor(max_workers=3) as executor:
            user_ids = list(range(1, 10))
            futures = [executor.submit(query_user_orders, uid) for uid in user_ids]
            
            for future in concurrent.futures.as_completed(futures):
                try:
                    orders = future.result()
                    print(f"查询到 {len(orders)} 条订单")
                except exception as e:
                    logger.error(f"并发查询失败: {e}")
    
    finally:
        pool.close_all()

if __name__ == "__main__":
    example_usage()

# 生产环境日志配置示例
import logging
from logging.handlers import rotatingfilehandler, timedrotatingfilehandler

# 按大小轮转的日志文件
size_handler = rotatingfilehandler(
    'mysql_operations.log',
    maxbytes=100*1024*1024,  # 100mb
    backupcount=10
)

# 按时间轮转的日志文件
time_handler = timedrotatingfilehandler(
    'mysql_slow_queries.log',
    when='midnight',  # 每天轮转
    backupcount=30
)

# 发送到监控系统（如elk）
# 可以使用logstash handler或直接api发送