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

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

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

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}; #输出到标准输出
}

# -*- coding: utf-8 -*-
import subprocess
from alive_progress import alive_bar # 非常丰富的进度条工具库
from tqdm import tqdm # 进度条工具库
import pandas as pd # 可以用来处理文本excel，csv等
from collections import OrderedDict
import argparse # 用来解析命令行参数 
import time
print('测试时间 : ', 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)