python多线程并发请求_python并发请求接口

再api测试时，避免不了高并发的测试情况。所以以下案例为线程并发请求代码，以请求百度为例

#!/usr/bin/env python
#!coding:utf-8
from __future__ import division
from threading import Thread
import requests
import matplotlib.pyplot as plt
import datetime
import time
import numpy as np
import json


class ThreadTest(Thread):
    def __init__(self, func, args=()):
        """
        :param func: 被测试的函数
        :param args: 被测试的函数的返回值
        """
        super(ThreadTest, self).__init__()
        self.func = func
        self.args = args

    def run(self):
        self.result=self.func(*self.args)

    def getResult(self):
        try:
            return self.result
        except BaseException as e:
            return e.args[0]


def faultinJection(code, seconds):
    """
    :param code: 状态码
    :param seconds: 请求响应时间
    :return:
    """
    url = "http://www.baidu.com/"
    r = requests.post(url=url)
    code = r.status_code
    seconds = r.elapsed.total_seconds()
    return code, seconds

def calculationTime(startTime,endTime):
    """计算两个时间之差，单位是秒"""
    return (endTime-startTime).seconds


def getResult(seconds):
    """获取服务端的响应时间信息"""
    data = {
      'Max': sorted(seconds)[-1],
      'Min': sorted(seconds)[0],
      'Median': np.median(seconds),
      '99%Line': np.percentile(seconds, 99),
      '95%Line': np.percentile(seconds, 95),
      '90%Line': np.percentile(seconds, 90)
    }
    return data


def highConcurrent(count):
    """
    对服务端发送高并发的请求
    :param count: 并发数
    :return:
    """
    startTime = datetime.datetime.now()
    sum = 0
    list_count = list()
    tasks = list()
    results = list()
    #失败的信息
    fails = []
    #成功任务数
    success = []
    codes = list()
    seconds = list()

    for i in range(1, count+1):
        t = ThreadTest(faultinJection, args=(i, i))
        tasks.append(t)
        t.start()

    for t in tasks:
        t.join()
        if t.getResult()[0] != 200:
            fails.append(t.getResult())
        results.append(t.getResult())

    endTime=datetime.datetime.now()
    for item in results:
        codes.append(item[0])
        seconds.append(item[1])
    for i in range(len(codes)):
        list_count.append(i)

    #生成可视化的趋势图
    # fig, ax = plt.subplots()
    # ax.plot(list_count, seconds)
    # ax.set(xlabel='number of times', ylabel='Request time-consuming',
    #        title='olap continuous request response time (seconds)')
    # ax.grid()
    # fig.savefig('olap.png')
    # plt.show()

    for i in seconds:
        sum += i
    rate = sum/len(list_count)
    totalTime = calculationTime(startTime=startTime, endTime=endTime)
    if totalTime < 1:
        totalTime = 1
    #吞吐量的计算
    try:
        throughput = int(len(list_count)/totalTime)
    except Exception as e:
        throughput = 0
        print(e.args[0])
    getResult(seconds=seconds)
    errorrate = "{}%".format(str(len(fails) / len(list_count) * 100))
    throughput = str(throughput)+'/S'
    timeData = getResult(seconds=seconds)
    dict1 = {
        '吞吐量': throughput,
        '平均响应时间': rate,
        '响应时间': timeData,
        '错误率': errorrate,
        '请求总数': len(list_count),
        '失败数': len(fails),
        '总共持续时间': "%0.3fs" %(float((endTime-startTime).total_seconds()))
    }
    return json.dumps(dict1, indent=True, ensure_ascii=False)

if __name__ == '__main__':
    print(highConcurrent(count=20))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

上面代码执行完之后会生成如下信息