基于k6和python进行自动化性能测试环球看热讯

标题

您现在在的位置是：首页>商业资讯 > 正文

基于k6和python进行自动化性能测试环球看热讯

博客园 2023-06-25 12:08:11

摘要：在性能测试中，达到相应的性能指标对于一个软件来说十分重要，在本文中，将介绍一种现代化性能测试工具k6。

本文分享自华为云社区《基于k6和python进行自动化性能测试》，作者：风做了云的梦。

(资料图)

当我们开发完成一个应用程序时，往往需要对其进行性能测试，以帮助我们更好的优化程序以及发现程序中的一些bug。在性能测试中，达到相应的性能指标对于一个软件来说十分重要，在本文中，将介绍一种现代化性能测试工具k6。

k6是一个开源工具，基于JavaScript可以编写k6的测试脚本，测试Web应用程序以及API的性能，支持HTTP等多种协议，可以很好地模拟各种高负载场景，充分验证程序稳定性和性能。k6支持Linux、MacOS等多个平台，通过k6官网根据提示即可在各个平台快速安装k6，终端输入k6 version出现如下显示说明安装成功。

以下是一个简单的k6测试脚本，通过k6的HTTP API模拟Get请求，并且休眠一秒钟：K

import http from "k6/http";import { sleep } from "k6";export default function () { http.get("https://test-api.com"); sleep(1);}

通过执行下面这行代码，运行脚本，即可对服务完成测试。

k6 run test-script.js

k6提供了丰富的功能，以下是k6常用的一些API，具体可以参考官网文档介绍：

- http.get(url, [options])：发送GET请求。- http.post(url, body, [options])：发送POST请求。- check(res, checks)：检查响应是否符合预期。- group(name, func)：将一组请求分组并统计性能指标。- sleep(duration)：休眠指定的时间。

k6的测试结果包括以下一些指标，可以根据这些指标，更好的优化程序。

- VUs：虚拟用户的数量。- Iterations：迭代次数。- RPS：每秒钟的请求数。- Duration：测试持续时间。- Data Sent/Received：发送和接收的数据量。- Checks：检查的数量。- Status codes：响应状态码的数量。- Errors：错误的数量。- Latency distribution：延迟分布。

通过Python和k6你可以更加高效的完成符合自己要求的自动化测试，Python可以提供非常多的工具库，用来收集处理k6返回的结果。我们可以编写以下k6测试脚本，并且通过Python去执行它，相关注释我已经标注出来，在handleSummary函数中，我们可以通过metrics来获取各种测试信息，具体如代码所示，可以参考官网关于metrics的介绍，同时自定义环境变量的使用也十分方便，可以参考代码中的使用方式。

import http from "k6/http";import { check, sleep} from "k6";import {Rate} from "k6/metrics";export default function() {    #post请求所需要的body体 let requestBody = { "xxx":[ "xxxxx"        ], "xxxx": __ENV.MyVar # MyVar为自定义的环境变量，可以通过__ENV调用，在执行脚本时可直接通过MyVar=xxx传值    };    #url const url = "http://example.com"; const payload = JSON.stringify(requestBody); const params = {    headers: { "Content-Type": "application/json",        },    timeout: "100s" #每个请求的超时时间    }; let res = http.post(url, payload, params);    #检测结果是否是200OK check(res, { "status is 200": (r) => r.status === 200 });}export function handleSummary(data) {        #通过data.metrics中的字段可以获取你想要的一些信息，例如每个请求的持续时间和吞吐量 const time = `${data.metrics.http_req_duration.values.avg.toFixed(3)}`; const rps = `${data.metrics.http_reqs.values.rate.toFixed(3)}`; const res = `${time} ${rps}`;         console.log(res); # 利用console.log可以将内容打印到控制台 return {stdout : res}; #输出到标准输出}

如下是一个Python代码示例，相关代码已经注释，通过Python中的subprocess模块执行k6脚本，并且捕获k6脚本的输出，通过pandas库进行整理输出到excel中。还可以通过argparse库解析命令行参数传入k6脚本中，更加灵活，高效。

# -*- coding: utf-8 -*-import subprocessfrom alive_progress import alive_bar # 非常丰富的进度条工具库from tqdm import tqdm # 进度条工具库import pandas as pd # 可以用来处理文本excel，csv等from collections import OrderedDictimport argparse # 用来解析命令行参数 import timeprint("测试时间 : ", time.strftime("%b %d %Y %H:%M:%S", time.gmtime(time.time())))print("************开始测试啦！ 祈祷不出错！**************")# 需要测试的测试语句集合test_examples = [ "aaaaaaa", "bbbbbbb", "ccccccc"]dataMap = {"test": test_examples}parser = argparse.ArgumentParser()parser.add_argument("-d", default="60s", help="duration time", dest="duration_time") #解析命令行参数，控制测试时间args = parser.parse_args()print("每条语句测试时间 : ", args.duration_time)vus = ["10", "20", "30", "40"] # 并发数集合 ，分别测试并发数为10,20,30,40的场景cols_name = ["1-avg/ms", "1-rps/s", "10-avg/ms", "10-rps/s","20-avg/ms", "20-rps/s","50-avg/ms", "50-rps/s"] # excel的列名# 循环测试，可以将多个需要测试的语句集合放入到dataMap中for (name, data) in dataMap.items():  print("当前测试的项目为 ：", name)        res = OrderedDict()        res["test_examples"] = [] for n in cols_name:                res[n] = []        df = pd.DataFrame(res) excel_name = name + ".xlsx" df.to_excel(excel_name, index=False) for query in data: print("当前测试语句为 ：", query)                origin = pd.read_excel(excel_name) with alive_bar(len(vus)) as bar: temp_dict = {} temp_dict["test_examples"] = query for vu in vus: keyRps = vu + "-rps/s" keyTime = vu + "-avg/ms" MyVar="MyVar=" + query #通过Popen执行k6脚本，并且捕获它的标准输出                                process = subprocess.Popen(["k6", "run", "--quiet", "script.js", "--env", MyVar, "--vus", vu, "--duration", args.duration_time], stdout=subprocess.PIPE, stderr=subprocess.PIPE)                                result = process.stdout.read()                                temp = result.split() temp_dict[keyTime] = temp[0].decode(); temp_dict[keyRps] = temp[1].decode(); print("并发：", vu, temp[0].decode(), temp[1].decode()) bar() #将脚本输出写到excel save_data = origin.append(temp_dict, ignore_index=True) save_data.to_excel(excel_name, index=False)

执行此Python脚本，可以得到类似以下输出：

1、k6官网文档链接：https://k6.io/docs/

2、k6安装链接：https://k6.io/docs/get-started/installation/