Python 使用多线程进行性能测试

只发送相同的请求，以及对请求是否成功进行判断，无需修改公共数据。所以没有实现队列和线程同步。

# 线程组列表
threads = []
# 异常请求列表
exception_list = []
# 成功请求返回结果列表
all_response_list = []
# 线程数
thread_number = 50
# 线程重复次数
repeat_time = 1

# set log的实现方法自行实现
def set_log(thread_name):
	pass

class MyThread(threading.Thread):
    def __init__(self, thread_id, loop_time):
        threading.Thread.__init__(self)
        self.thread_id = thread_id
        self.loop_time = loop_time

    def run(self):
        for i in range(0, self.loop_time):
            # 获取接口response
            try:
                single_request_start_time_count = time.perf_counter()

                set_log(thread_name=str(self.thread_id) + f'_{i + 1}')

                single_request_end_time_count = time.perf_counter()
                # 计算单词request耗时
                single_request_spend_time = single_request_end_time_count - single_request_start_time_count

                # 输出线程id，单次request到response耗时
                single_response = [str(self.thread_id) + f'_{i + 1}', str(single_request_spend_time)]
                # 将单词输入加入列表
                all_response_list.append(single_response)

            # 对返回异常进行处理
            except Exception:
                exc_value = sys.exc_info()[1]
                actual = str(exc_value)
                # 测试结果写入EXCEL
                exception_list.append(actual)


if __name__ == '__main__':
    # 多线程开始时间
    start_time_count = time.perf_counter()

    for thread_name in range(1, thread_number + 1):
        thread = MyThread(thread_name, loop_time=repeat_time)
        thread.start()
        threads.append(thread)

    for t in threads:
        t.join()

    print('*********')
    # 多线程结束时间
    end_time_count = time.perf_counter()
    # 多线程执行时间
    spend_time = end_time_count - start_time_count

    print(f'多线程共执行了{spend_time}秒')
    print(f'多线程共有{thread_number}个线程')
    print(f'多线程共进行了{repeat_time}次循环')
    print(f'多线程执行了{thread_number * repeat_time}个多线程任务')
    print(f'多线程共{len(all_response_list)}个多线程任务响应')

    exception_number = len(exception_list)
    print(f'多线程任务异常总数是{exception_number}')
    print(f'多线程任务异常清单是{exception_list}')

    spend_time_list = []

    for result in all_response_list:
        spend_time_list.append(result[1])

    save.save_log_response(all_response_list)

    print(f'单次最大执行时间{max(spend_time_list)}秒')
    print(f'单次最小执行时间{min(spend_time_list)}秒')
    total_spend_time = 0
    for single_spend_time in spend_time_list:
        total_spend_time += float(single_spend_time)

    avg_time = total_spend_time / (thread_number * repeat_time)
    print(f'单次平均执行时间{avg_time}秒')

    tps = (thread_number * repeat_time) / spend_time
    print(f'TPS为{tps}')