Python并发（多线程）测试MySQL数据库性能

以下脚本主要是测试并发执行insert操作MySQL的性能。

创建表

create database if not exists data;
use data

create table if not exists tbl_player(
    id int unsigned auto_increment,
    title varchar(100) not null,
    author varchar(40) not null,
    write_date datetime,
    primary key (id)
);

Python3 脚本

import time
import pymysql  # pip3 install PyMySQL
import threading
from time import ctime

row_count = 100000000000  # 总insert的数据量
threads_count = 50  # 并发线程数


def time_me(fn):
    """
    记录方法执行时间
    :param args:
    :param kwargs:
    :return:
    """

    def _wrapper(*args, **kwargs):
        start = time.time()
        fn(*args, **kwargs)
        seconds = time.time() - start
        print("{func}函数每{count}条数数据写入耗时{sec}秒".format(func='ordinary_insert', count=args[0], sec=seconds))

    return _wrapper


@time_me
def run_insert(count):
    db = pymysql.connect(host="127.0.0.1", port=3306, user="root", passwd="1", db="data", charset="utf8")
    cur = db.cursor()
    for i in range(count):
        sql = "insert into tbl_player (title, author, write_date) values ('MySQL测试', 'MySQL测试', NOW());"
        try:
            cur.execute(sql)
            db.commit()  # 每次都提交
        except Exception as e:
            print(e)
    db.close()  # 关闭连接


local_var = threading.local()


def Clean(args):
    local_var.name = args
    run_insert(int(row_count / threads_count))


threads = []
for i in range(threads_count):
    t = threading.Thread(target=Clean, args=(i,))
    threads.append(t)

print('start:', ctime())
start = time.time()

if __name__ == '__main__':
    for i in threads:
        i.start()
    for i in threads:
        i.join()

seconds = time.time() - start
print('end:', ctime())
print("{func}函数每{count}条数数据写入耗时{sec}秒".format(func='ordinary_insert', count=row_count, sec=seconds))