# 檢查MySQL是否安裝
mysql --version

# 如果未安裝，使用apt安裝（其他系統(tǒng)請(qǐng)參考官方文檔）
sudo apt update
sudo apt install mysql-server mysql-client

# 啟動(dòng)MySQL服務(wù)
sudo systemctl start mysql
sudo systemctl enable mysql

# 安全初始化（設(shè)置root密碼等）
sudo mysql_secure_installation

# 創(chuàng)建虛擬環(huán)境（推薦）
python -m venv mysql-monitor-env
source mysql-monitor-env/bin/activate  # Linux/Mac
# 或 mysql-monitor-env\Scripts\activate  # Windows

# 安裝pymysql
pip install pymysql

# 安裝額外的工具庫(kù)，用于后續(xù)的監(jiān)控?cái)?shù)據(jù)可視化
pip install matplotlib pandas

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

def create_test_database():
    """創(chuàng)建測(cè)試數(shù)據(jù)庫(kù)和表，并插入模擬數(shù)據(jù)"""
    
    # 連接MySQL（請(qǐng)?zhí)鎿Q為你的實(shí)際密碼）
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='your_password',  # 改成你的MySQL root密碼
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor
    )
    
    try:
        with connection.cursor() as cursor:
            # 創(chuàng)建數(shù)據(jù)庫(kù)
            cursor.execute("CREATE DATABASE IF NOT EXISTS ecommerce_monitor")
            cursor.execute("USE ecommerce_monitor")
            
            # 創(chuàng)建用戶表
            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)
                )
            """)
            
            # 創(chuàng)建訂單表（故意不加索引，用于演示慢查詢）
            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上加索引！
                )
            """)
            
            # 插入測(cè)試用戶數(shù)據(jù)
            print("插入用戶數(shù)據(jù)...")
            users = []
            for i in range(1, 1001):  # 1000個(gè)用戶
                users.append((f'user{i}', f'user{i}@example.com'))
            
            cursor.executemany(
                "INSERT INTO users (username, email) VALUES (%s, %s)",
                users
            )
            
            # 插入測(cè)試訂單數(shù)據(jù)（更多數(shù)據(jù)，用于模擬真實(shí)場(chǎng)景）
            print("插入訂單數(shù)據(jù)...")
            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)
                
                # 隨機(jī)狀態(tài)
                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("測(cè)試數(shù)據(jù)創(chuàng)建完成！")
            
    finally:
        connection.close()

if __name__ == "__main__":
    create_test_database()

# 編輯MySQL配置文件
sudo nano /etc/mysql/mysql.conf.d/mysqld.cnf  # Ubuntu路徑，其他系統(tǒng)可能不同

# 在[mysqld]部分添加或修改以下配置：
"""
[mysqld]
# 開啟慢查詢?nèi)罩?
slow_query_log = 1

# 指定慢查詢?nèi)罩疚募窂?
slow_query_log_file = /var/log/mysql/mysql-slow.log

# 設(shè)置慢查詢閾值（單位：秒），這里設(shè)為1秒，生產(chǎn)環(huán)境通常設(shè)2-3秒
long_query_time = 1

# 記錄未使用索引的查詢（即使執(zhí)行時(shí)間沒超過閾值）
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():
    """執(zhí)行一些會(huì)觸發(fā)慢查詢的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("執(zhí)行可能較慢的查詢...")
            
            # 查詢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（無索引條件查詢）耗時(shí): {elapsed1:.3f}秒，返回 {len(result1)} 條記錄")
            
            # 查詢2：復(fù)雜聯(lián)表查詢
            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（復(fù)雜聯(lián)表分組）耗時(shí): {elapsed2:.3f}秒")
            
            # 查詢3：使用索引的快速查詢（作為對(duì)比）
            start = time.time()
            cursor.execute("SELECT * FROM users WHERE username = 'user500'")
            result3 = cursor.fetchall()
            elapsed3 = time.time() - start
            print(f"查詢3（使用索引查詢）耗時(shí): {elapsed3:.3f}秒")
            
    finally:
        connection.close()

if __name__ == "__main__":
    generate_slow_queries()

# 查看慢查詢?nèi)罩荆ㄐ枰猻udo權(quán)限）
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():
    """根據(jù)慢查詢分析添加缺失的索引"""
    
    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("索引添加完成！")
            
            # 驗(yàn)證優(yōu)化效果
            print("\n驗(yàn)證優(yōu)化效果：")
            cursor.execute("EXPLAIN SELECT * FROM orders WHERE user_id = 500 AND created_at > '2023-01-01'")
            explain_result = cursor.fetchone()
            
            print(f"查詢執(zhí)行計(jì)劃：")
            print(f"- 使用的索引: {explain_result.get('key', '無')}")
            print(f"- 掃描行數(shù): {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性能監(jiān)控器"""
    
    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):
        """獲取慢查詢摘要統(tǒng)計(jì)"""
        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):
        """獲取連接統(tǒng)計(jì)"""
        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):
        """獲取表訪問統(tǒng)計(jì)"""
        with self.connection.cursor() as cursor:
            # 注意：這個(gè)查詢需要開啟某些consumer，默認(rèn)可能沒有數(shù)據(jù)
            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):
        """監(jiān)控循環(huán)，定期收集性能數(shù)據(jù)"""
        print(f"開始性能監(jiān)控，每{interval}秒采樣一次，持續(xù){duration}秒...")
        
        data_points = []
        start_time = time.time()
        
        while time.time() - start_time < duration:
            timestamp = datetime.now()
            
            # 收集各種性能指標(biāo)
            slow_queries = self.get_slow_queries_summary()
            connections = self.get_connection_stats()
            
            # 記錄數(shù)據(jù)點(diǎn)
            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}] 慢查詢數(shù): {data_point['slow_query_count']}, "
                  f"活躍連接: {data_point['active_connections']}")
            
            time.sleep(interval)
        
        # 轉(zhuǎn)換為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("=== 當(dāng)前慢查詢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"   平均耗時(shí): {query['avg_time_sec']:.3f}s, 執(zhí)行次數(shù): {query['exec_count']}")
        
        print("\n=== 當(dāng)前連接統(tǒng)計(jì) ===")
        connections = monitor.get_connection_stats()
        for conn in connections:
            print(f"用戶: {conn['user']}, 連接數(shù): {conn['connection_count']}")
        
        # 運(yùn)行監(jiān)控循環(huán)（生產(chǎn)環(huán)境可以改為后臺(tái)任務(wù)）
        # df = monitor.monitor_loop(interval=10, duration=60)
        # print(f"\n監(jiān)控?cái)?shù)據(jù)摘要:\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"連接池初始化完成，最大連接數(shù): {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):
        """創(chuàng)建新連接"""
        try:
            conn = pymysql.connect(**self.connection_args)
            
            # 設(shè)置連接屬性，便于追蹤
            with conn.cursor() as cursor:
                cursor.execute("SET @python_client_id = %s", 
                             (f"pool_conn_{len(self.pool)}",))
            
            logger.debug(f"創(chuàng)建新數(shù)據(jù)庫(kù)連接: {conn.server_version}")
            return conn
        except Exception as e:
            logger.error(f"創(chuàng)建數(shù)據(jù)庫(kù)連接失敗: {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"連接健康檢查失敗，關(guān)閉異常連接: {e}")
                    try:
                        conn.close()
                    except:
                        pass
            
            # 補(bǔ)充連接
            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"健康檢查完成，活躍連接數(shù): {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"獲取數(shù)據(jù)庫(kù)連接超時(shí)，等待{wait_time:.2f}秒后仍無可用連接")
            raise Exception("數(shù)據(jù)庫(kù)連接池耗盡")
        
        try:
            # 記錄連接獲取
            logger.debug(f"獲取數(shù)據(jù)庫(kù)連接，當(dāng)前使用中: {len(self.in_use)}")
            
            # 執(zhí)行查詢前的準(zhǔn)備
            with conn.cursor() as cursor:
                cursor.execute("SET @query_start_time = NOW(6)")
            
            yield conn
            
        except pymysql.Error as e:
            # 記錄數(shù)據(jù)庫(kù)錯(cuò)誤
            error_code, error_msg = e.args
            logger.error(f"數(shù)據(jù)庫(kù)操作錯(cuò)誤 [{error_code}]: {error_msg}")
            
            # 根據(jù)錯(cuò)誤類型決定是否關(guān)閉連接
            if error_code in (2006, 2013):  # 連接相關(guān)錯(cuò)誤
                logger.warning("連接異常，將關(guān)閉并創(chuàng)建新連接")
                try:
                    conn.close()
                except:
                    pass
                conn = self._create_connection()
            raise
            
        finally:
            # 記錄查詢執(zhí)行時(shí)間
            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"慢查詢檢測(cè): 執(zhí)行時(shí)間 {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"歸還數(shù)據(jù)庫(kù)連接，執(zhí)行時(shí)間: {exec_time:.3f}秒")
    
    def close_all(self):
        """關(guān)閉所有連接"""
        with self.lock:
            for conn in self.pool:
                try:
                    conn.close()
                except:
                    pass
            self.pool.clear()
            self.in_use.clear()
        
        logger.info("連接池已關(guān)閉所有連接")

# 使用示例
def example_usage():
    """使用帶監(jiān)控的連接池示例"""
    
    # 創(chuàng)建連接池
    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"訂單總數(shù): {result['count']}")
        
        # 示例2：事務(wù)操作
        with pool.get_connection() as conn:
            try:
                with conn.cursor() as cursor:
                    # 開始事務(wù)
                    cursor.execute("START TRANSACTION")
                    
                    # 插入新訂單
                    cursor.execute(
                        "INSERT INTO orders (user_id, amount, status) VALUES (%s, %s, %s)",
                        (1, 99.99, 'pending')
                    )
                    
                    # 更新用戶統(tǒng)計(jì)（模擬業(yè)務(wù)邏輯）
                    cursor.execute(
                        "UPDATE users SET email = %s WHERE id = %s",
                        ('updated@example.com', 1)
                    )
                    
                    # 提交事務(wù)
                    conn.commit()
                    logger.info("事務(wù)提交成功")
                    
            except Exception as e:
                conn.rollback()
                logger.error(f"事務(wù)回滾: {e}")
                raise
        
        # 示例3：批量查詢（模擬業(yè)務(wù)高峰）
        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()
        
        # 模擬并發(fā)查詢
        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"并發(fā)查詢失敗: {e}")
    
    finally:
        pool.close_all()

if __name__ == "__main__":
    example_usage()