二、多并发实现接口压力测试

一、flsak接口压力测试

import base64
import logging
import os, cv2,time
import urllib, glob
import numpy as np
import requests, time, json, threading, random
import tracebackdef cv_imread(file_path):# 读取中文pathcv_img = cv2.imdecode(np.fromfile(file_path, dtype=np.uint8), -1)return cv_imgclass LogTxt:def __init__(self, txt_path):self.current_path = txt_pathcreate_time = time.strftime("%Y-%m-%d %H:%M:%S")self.txt_handle = open(self.current_path, 'a',encoding="utf-8")self.info(f" ----------- start test in {create_time} --------\n")def info(self, content):try:now_time = time.strftime("%Y-%m-%d %H:%M:%S")write_content = now_time + f' - keyPoint - INFO - :' + contentself.txt_handle.write(write_content + ' \n')self.txt_handle.flush()print(write_content)except:print("=============== > LogTxt info have bug < =========== ", traceback.format_exc())def close(self):self.txt_handle.close()class Presstest(object):"""并发压力测试"""def __init__(self, press_url, file,logger):self.press_url = press_urlself.file = fileself.logger=loggerdef test_interface(self):'''压测接口'''global INDEXINDEX += 1global ERROR_NUMglobal TIME_LENStry:start = time.time()logger.info(f"识别本次高清图开始时间为：{start}")r = requests.post(self.press_url, files=self.file)# print(r.text)logger.info(f"识别本次高清图开始结果为：{r.text}")end = time.time()total_time=end-startlogger.info(f"识别本次高清图结束时间为：{start}")logger.info(f"识别本次高清图共计时间为：{total_time}")TIME_LENS.append(end - start)logger.info("高清图识别结束，进行下一张高清图识别。\n\n")print('end')except Exception as e:ERROR_NUM += 1logger.info(f"识别本次高清图发生异常为：{e}")print(e)# headers = {'Content-Type': 'application/json; charset=UTF-8',#            'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0'}def test_onework(self):'''一次并发处理单个任务'''i = 0while i < ONE_WORKER_NUM:i += 1self.test_interface()# print('one worker num {}'.format(i))logger.info('one worker num {}'.format(i))time.sleep(LOOP_SLEEP)##---------------------------------------------# todo 通用http获取webapi请求结果方法## --------------------------------------------# def do_request(self, press_url,register_data):#     '''通用http获取webapi请求结果方法'''# headers = {#     'Content-Type': 'application/json; charset=UTF-8',# }# request = urllib.request.Request(url, json.dumps(payload).encode("utf-8"), headers=headers)# retry_num = 0# while retry_num < 3:#     response = urllib.request.urlopen(request, timeout=300)#     if not response or response.status == 421:#         time.sleep(1)#         retry_num = retry_num + 1#         continue#     else:#         break# response_content = response.read()# if hasattr(response_content, 'decode'):#     response_content = response_content.decode('utf-8')## return response_contentdef run(self):'''使用多线程进程并发测试'''t1 = time.time()Threads = []for i in range(THREAD_NUM):# print('thread Num {}'.format(i))logger.info('thread Num {}'.format(i))t = threading.Thread(target=self.test_onework, 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)# print(INDEX)logger.info("===============压测结果===================")logger.info("URL: {}".format(self.press_url))logger.info("任务数量:{} * {} = {}".format(THREAD_NUM,ONE_WORKER_NUM,THREAD_NUM * ONE_WORKER_NUM))logger.info("总耗时(秒):{}".format(t2-t1))logger.info("每次请求耗时(秒):{}".format((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))logger.info("每秒承载请求数:{}".format(1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))))logger.info("错误数量:{}".format(ERROR_NUM))logger.info("{}".format(INDEX))logger.info("----------------本次测试结束，以上为测试结果-----------------")if __name__ == '__main__':press_url = 'http://192.168.2.93:8095/recog'TIME_LENS = []INDEX = 0THREAD_NUM = 25 # 并发线程总数ONE_WORKER_NUM = 50  # 每个线程的循环次数LOOP_SLEEP = 0  # 每次请求时间间隔(秒)ERROR_NUM = 0  # 出错数logger=LogTxt(rf"./log/thread-{THREAD_NUM}.log")path1 = r"C:\val"for path in glob.glob(os.path.join(path1, "*.jpg" or "*.png")):# path=r"/data2/enducation/answer_card/answer-card-recognition/pic/2021-08-17_14_16_18.jpg"with open(path, "rb") as f:img_data = f.read()  # <class 'bytes'># print(type(img_data))files = {'file': img_data}obj = Presstest(press_url, files,logger)obj.run()print('onetime')

二、多并发实现接口压力测试相关推荐

学习笔记－JMeter 进行接口压力测试
一.压力测试场景设置 1.场景设定:进行接口压力测试时,有单场景也有混合场景.单场景就是对一个接口进行请求:混合场景需要对多个接口进行请求,在流程类业务场景会运用到 2.压测时间设定:通常时间为10- ...
用python做简单的接口压力测试
一个多月没有更博了,最近研究了一下接口的压力测试,主要来说就是连续频繁的对接口的调用,来测试接口的响应速度.返回结果,找到接口的性能瓶颈,最大承受极限等. 做接口压力测试的方法和工具很多,比较常用的工 ...
从零开始的Nginx [ 8 ] --- nginx 的性能优化：ab接口压力测试工具，tomcat企业运维，WEB站点部署，项目上线
文章目录 nginx 性能优化 1.当前系统结构瓶颈 2.了解业务模式 3.性能与安全 4.系统与nginx性能优化 1.文件句柄 2.设置方式 3.系统全局性修该和用户局部性修改 4.进程局部性修改 ...
【Jmeter基础篇】03：如何进行post接口压力测试
目前网络上有很多很多Jmeter的压力测试使用指南,但是在实际使用过程中,很多朋友会遇到许多并不处于理想状态的情况,比如某些参数不知道应该往哪里放,具体怎么分析测试结果等等. 所谓压力测试,实质就是让 ...
Jmeter验证码图片识别注册接口压力测试实战
写在前面前段时间写了一篇博文<Jmeter验证码注册接口压力测试实战>,前两天又登录对应的网站看了一下.发现原来滑动验证的方式改成了图片验证码的方式,防止用户多次发送手机验证码. 看到这 ...
Jmeter验证码注册接口压力测试实战
主要学习的内容包括发送HTTP请求.HTTP信息头管理器.HTTP cookies管理器.用户定义的变量.响应断言.CSV数据文件设置的使用. 测试网址及测试接口等说明提供的测试网址为: http: ...
Postman 接口压力测试【嘿】
Postman 接口压力测试一.准备二.压力测试步骤三.导出csv文件一.准备 1.1 下载Postman 软件 postman下载https://www.postman.com/ 1.2 首 ...
接口压力测试：Postman【Postman通常用于做接口测试，同时也可以用于作为压力测试】、Jmeter【专门做压力测试】、Loadrunner、Apache AB、Webbench
做开发的同学一定会遇到接口对接,今天介绍两个对接测试两个我个人认为比较好的测试工具 postman和jmeter 1.postman通常用于做接口测试,同时也可以用于作为压力测试 2.jmeter做压 ...
python分析股票支撑压力_python+requests接口压力测试500次,查看响应时间的实例
接口压力测试500次,查看响应时间 import json import requests import logging logging.basicConfig(level=logging.INFO, ...

二、多并发实现接口压力测试

一、flsak接口压力测试

二、多并发实现接口压力测试相关推荐

最新文章

热门文章