当前位置: article > 正文

Python接口并发压力测试（单接口，多接口参数化）+异步aiohttp_压测单个api需要配置多个参数吗

作者：不正经 | 2024-03-09 00:03:19

踩

压测单个api需要配置多个参数吗

一单接口

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import json
import random
import datetime

import requests
import threading
import time


class Presstest(object):
    headers = {
        'Content-Type': 'application/json; charset=UTF-8'
    }

    def __init__(self, login_url, press_url, phone, password):
        self.login_url = login_url
        self.press_url = press_url
        self.phone = phone
        self.password = password
        self.session = requests.Session()
        self.session.headers = self.headers

    def login(self):
        '''登陆获取session'''
        data = {"mobile": self.phone, "pwd": self.password, "type": "1"}
        res = self.session.post(self.login_url, data=json.dumps(data))
        XToken = res.json().get('data').get('token')
        self.session.headers['token'] = XToken

    def testinterface(self):
        '''压测接口'''
        # self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
        data = {"id": 418}
        global ERROR_NUM
        try:
            print('开始调接口：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
            response = self.session.post(self.press_url, data=json.dumps(data))
            if response.json().get('code') != 0:
                print(response.json())
                ERROR_NUM += 1
        except Exception as e:
            print(e)
            ERROR_NUM += 1

    def testonework(self):
        '''一次并发处理单个任务'''
        i = 0
        while i < ONE_WORKER_NUM:
            i += 1
            self.testinterface()
        time.sleep(LOOP_SLEEP)

    def run(self):
        '''使用多线程进程并发测试'''
        t1 = time.time()
        Threads = []

        for i in range(THREAD_NUM):
            t = threading.Thread(target=self.testonework, name="T" + str(i))
            t.setDaemon(True)
            Threads.append(t)

        for t in Threads:
            t.start()
        for t in Threads:
            t.join()
        t2 = time.time()

        print("===============压测结果===================")
        print("URL:", self.press_url)
        print("任务数量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
        print("总耗时(秒):", t2 - t1)
        print("每次请求耗时(秒):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
        print("每秒承载请求数:", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
        print("错误数量:", ERROR_NUM)


if __name__ == '__main__':
    login_url = 'http://test/sale/login/login'
    press_url = 'http://test/sale/hsOrder/afterExamineAdopt'
    phone = "150000000"
    password = "123456"

    THREAD_NUM = 1  # 并发线程总数
    ONE_WORKER_NUM = 5  # 每个线程的循环次数
    LOOP_SLEEP = 0.1  # 每次请求时间间隔(秒)
    ERROR_NUM = 0  # 出错数

    obj = Presstest(login_url=login_url, press_url=press_url, phone=phone, password=password)
    obj.login()
    obj.run()

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

二多接口参数化（实现多接口参数化并发，data和url必须一一对应，且THREAD_NUM并发线程数不能大于url_list长度）

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import json
import random
import datetime

import requests
import threading
import time


class Presstest(object):
    headers = {
        'Content-Type': 'application/json; charset=UTF-8'
    }

    def __init__(self, login_url, press_url, phone, password):
        self.login_url = login_url
        self.press_url = press_url
        self.phone = phone
        self.password = password
        self.session = requests.Session()
        self.session.headers = self.headers

    def login(self):
        '''登陆获取session'''
        data = {"mobile": self.phone, "pwd": self.password, "type": "1"}
        res = self.session.post(self.login_url, data=json.dumps(data))
        XToken = res.json().get('data').get('token')
        self.session.headers['token'] = XToken

    def testinterface(self):
        '''压测接口'''
        # self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
        data = {"id": 418}
        global ERROR_NUM
        try:
            print('开始调接口111111：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
            response = self.session.post(self.press_url, data=json.dumps(data))
            if response.json().get('code') != 0:
                print(response.json())
                ERROR_NUM += 1
        except Exception as e:
            print(e)
            ERROR_NUM += 1

    def testinterface2(self, url, data):
        '''压测接口'''
        # self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
        # data = {"orderId": 1179, "cause": "让人"}
        global ERROR_NUM
        try:
            print('开始调接口2222222：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
            print("接口请求入参url，data==", url, data)
            response = self.session.post(url,
                                         data=json.dumps(data))
            if response.json().get('code') != 0:
                print(response.json())
                ERROR_NUM += 1
        except Exception as e:
            print(e)
            ERROR_NUM += 1

    def testonework(self, url, data):
        '''一次并发处理单个任务'''
        i = 0
        while i < ONE_WORKER_NUM:
            i += 1
            self.testinterface2(url, data)
            self.testinterface()
        time.sleep(LOOP_SLEEP)

    def run(self):
        '''使用多线程进程并发测试'''
        t1 = time.time()
        Threads = []
        # 实现多接口参数化并发，data和url必须一一对应，且THREAD_NUM并发线程数不能大于url_list长度
        data_list = [{"orderId": 1194, "cause": "让人"},
                     {"orderId": 1193, "cause": "让人"},
                     {"orderId": 1192, "cause": "让人"}]
        url_list = ["http://test/api1",
                    "http://test/api2",
                    "http://test/api3"]
        list_arr = list(range(0, len(data_list)))
        print("index========", list_arr)
        for i in range(THREAD_NUM):
            index = random.choice(list_arr)
            list_arr.remove(index)
            t = threading.Thread(target=self.testonework(url_list[index], data_list[index]), name="T" + str(i))
            t.setDaemon(True)
            Threads.append(t)

        for t in Threads:
            t.start()
        for t in Threads:
            t.join()
        t2 = time.time()

        print("===============压测结果===================")
        print("URL:", self.press_url)
        print("任务数量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
        print("总耗时(秒):", t2 - t1)
        print("每次请求耗时(秒):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
        print("每秒承载请求数:", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
        print("错误数量:", ERROR_NUM)


if __name__ == '__main__':
    login_url = 'http://test/login'
    press_url = 'http://test/afterExamineAdopt'
    phone = "1500000000"
    password = "123456"

    THREAD_NUM = 3  # 并发线程总数
    ONE_WORKER_NUM = 1  # 每个线程的循环次数
    LOOP_SLEEP = 0  # 每次请求时间间隔(秒)
    ERROR_NUM = 0  # 出错数

    obj = Presstest(login_url=login_url, press_url=press_url, phone=phone, password=password)
    obj.login()
    obj.run()

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

三多接口并发调用方法二

import datetime
import json

import requests
import threading
import time


def post_request(url, data):
    start_time = time.time()
    print('开始调接口111111：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
    response = requests.post(url, data=json.dumps(data), headers={"Content-Type": "application/json;charset=UTF-8"})
    print('调用结束：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
    end_time = time.time()
    duration = end_time - start_time
    print("Response from", url, ":", data, response.text)
    print("Request duration:", duration * 1000, "ms")


data = [{"orderId": 1194, "cause": "让人"},
        {"orderId": 1193, "cause": "让人"},
        {"orderId": 1192, "cause": "让人"}]
urls = ["http://test/api1",
        "http://test/api2",
        "http://test/api3"]

threads = []
for i in range(len(urls)):
    t = threading.Thread(target=post_request, args=(urls[i], data[i]))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

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

四多接口同时并发（相当于集合点）（异步实现集合点）

import asyncio
import datetime
import aiohttp
async def make_request(session, url, data):
    print('开始调接口：', datetime.datetime.now().strftime('%Y-%m-%d- %H:%M:%S:%f'))
    async with session.post(url, json=data) as response:
        result = await response.json(content_type='text/html', encoding='utf-8')
        return result


async def run_concurrent_requests(urls, datas, max_concurrent_requests):
    headers = {
        "Content-Type": "application/json;charset=UTF-8"
    }
    tasks = []
    async with aiohttp.ClientSession() as session:
        session.headers = headers
        session.headers["token"] = "BEE010F2A6D0696DD90A82FF28B21AF2"
        sem = asyncio.Semaphore(max_concurrent_requests)
        list_arr = list(range(0, len(datas)))
        for index in list_arr:
            await sem.acquire()
            task = asyncio.ensure_future(make_request(session, urls[index], datas[index]))
            task.add_done_callback(lambda t: sem.release())
            tasks.append(task)
        responses = await asyncio.gather(*tasks)
        return responses


if __name__ == '__main__':
    urls = ['http://test/SubmitSettlement'] * 5
    datas = [{"idArr": [2086]}, {"idArr": [2090]}, {"idArr": [2089]}, {"idArr": [2086]}, {"idArr": [2090]}]
    max_concurrent_requests = 5
    loop = asyncio.get_event_loop()
    responses = loop.run_until_complete(run_concurrent_requests(urls, datas, max_concurrent_requests))
    print(responses)
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

在这里插入图片描述

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】