性能検証用のテストコード

import pdb
import time
from threading import Event, Thread

import mysql.connector
import psutil

# TODO:同時に接続切断が発生した場合
# TODO:動作用のDBダンプ

# pip install psutil
# install mysql-connector-python

# DOS窓からMySQLの再起動（DOS窓は管理者で立ち上げること）
# sc stop MySQL57 & & timeout 2 & & sc start MySQL57


def monitor():
    """CPU使用率とかのモニタ用
    """
    import sys
    cnt = 0
    cpu = 0
    max_used = 0
    interval = 0.1
    min_used = sys.maxsize
    while not event.wait(interval):
        _ = psutil.cpu_percent()
        cpu = cpu + _

        total, available, percent, used, free = psutil.virtual_memory()
        if(used > max_used):
            max_used = used
        if(used < min_used):
            min_used = used
        # print(f"CPU:{_}, MEMORY:{psutil.virtual_memory()}")
        cnt = cnt + 1
    _ = round(cpu/cnt, 1)
    __ = "{:,}".format(max_used)
    ___ = "{:,}".format(min_used)
    ____ = "{:,}".format(max_used-min_used)
    print(f"monitor_count:{cnt}, monitor_interval: {interval}")
    print(
        f"cpu_average: {_}, mem_max_used: {__}, mem_min_used: {___}, max-min: {____}")


def deco_monitor(func):
    """モニタのデコレータ
    """
    def wrapper(*args, **kwargs):
        start = time.time()
        event.clear()
        Thread(target=monitor).start()
        func(*args, **kwargs)
        event.set()
        elapsed_time = time.time() - start

        _ = round(elapsed_time, 2)
        __ = round(elapsed_time * 1000/loop, 4)
        print(f"実行時間(s): {_}, 実行時間/{loop}(ms): {__}")
        time.sleep(1)
    return wrapper


def info():
    """MySQLの情報表示
    """
    print()
    connection = mysql.connector.connect(
        host='127.0.0.1', user='root', password='password', database='test')
    cursor = connection.cursor()

    # processlist表示
    cursor.execute("show processlist")
    for record in cursor:
        print(f"processlist: {record}")

    # threadの使用状況表示
    cursor.execute("show status like 'Threads%'")
    list = []
    for record in cursor:
        list.append(record)

    cursor.execute("show variables like 'thread_cache_size'")
    for record in cursor:
        list.append(record)

    print(f"threadInfo: {list}")

    connection.close()
    print()


def header(title):
    print()
    print("--------------------------------------------------")
    print(title)
    print()


@deco_monitor
def 何もしてない時(sec):
    header(f"何もしてない（平常時の負荷確認用）")
    info()
    time.sleep(sec)


connect_count = 0


@deco_monitor
def 接続切断のみ():
    header(f"接続切断だけを{loop}回実行する。")
    info()
    global connect_count
    connect_count = 0
    for cnt in range(loop):
        connection = mysql.connector.connect(
            host='127.0.0.1', user='root', password='password', database='test')
        connect_count = connect_count + 1
        connection.close()


@deco_monitor
def SQL実行_都度接続():
    header(f"SQL1つを{loop}回実行する。（都度接続）")
    info()
    global connect_count
    connect_count = 0
    for cnt in range(loop):
        connection = mysql.connector.connect(
            host='127.0.0.1', user='root', password='password', database='test')
        connect_count = connect_count + 1
        cursor = connection.cursor()
        cursor.execute("select * from hoge")

        for record in cursor:
            record

        connection.close()


@deco_monitor
def SQL実行_接続1回だけ():
    header(f"SQL1つを{loop}回実行する＠コネクションプール使用。（接続自体は1回）")
    info()
    global connect_count
    connect_count = 0
    for cnt in range(loop):
        connection = mysql.connector.connect(
            host='127.0.0.1', user='root', password='password', database='test', pool_size=1)
        connect_count = connect_count + 1
        cursor = connection.cursor()
        cursor.execute("select * from hoge")

        for record in cursor:
            record

        connection.close()


import os

event = Event()
monitor_th = Thread(target=monitor)
loop = 5000
try:
    # MySQLの再起動（win用）
    # os.system(
    #     "sc stop MySQL57 >nul && timeout 3 >nul && sc start MySQL57 >nul && timeout 1 >nul")
    何もしてない時(10)
    接続切断のみ()
    SQL実行_都度接続()
    SQL実行_接続1回だけ()
finally:
    event.set()
    print(f"connect_count: {connect_count}")