locust 当做库使用，通过代码启动

使用Environment

import json
import gevent
from gevent._semaphore import Semaphore
from locust import FastHttpUser, task, between, events
from locust.env import Environment
from locust.html import get_html_report
from locust.stats import stats_printer, stats_history

all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()

def on_hatch_complete(**kwargs):
    # 创建钩子方法
    all_locusts_spawned.release()
    # 挂在到locust钩子函数（所有的Locust示例产生完成时触发）
    events.spawning_complete.add_listener(on_hatch_complete)

class MyHttpUser(FastHttpUser):
    wait_time = between(0.5, 2)  # 执行任务 等待时长   检查点 思考时间
    host = 'http://127.0.0.1'

    # def on_start(self):
    #     # 所有任务启动前执行
    #     all_locusts_spawned.wait()
    #     print('所有任务启动前执行')

    @task(1)
    def login(self):
        url = 'https://www.zaojingyoutu.top:8000/api/login/'
        payload = json.dumps({'username': 'zjy', 'password': '123456'})
        headers = {'Content-Type': 'application/json'}
        # all_locusts_spawned.wait()
        self.client.post(url, data=payload, headers=headers)


env = Environment(user_classes=[MyHttpUser], events=events)
runner = env.create_local_runner()
# start a WebUI instance
web_ui = env.create_web_ui("127.0.0.1", 8089)


# 执行初始化事件处理程序(只有在注册了任何事件时才真正需要)
env.events.init.fire(environment=env, runner=runner, web_ui=web_ui)
# 启动一个绿色程序，周期性地输出当前的统计信息
gevent.spawn(stats_printer(env.stats))
# 启动一个greenlet，将当前统计数据保存到历史记录中
gevent.spawn(stats_history, env.runner)
# start the test
runner.start(2, spawn_rate=10)


# in 60 seconds stop the runner
# gevent.spawn_later(60, lambda: runner.quit())    #直接关闭
gevent.spawn_later(60, lambda: runner.stop())

# wait for the greenlets
runner.greenlet.join()
res=get_html_report(env)
with open('report.html', 'w') as f:
    f.write(res)
# 关闭web界面
# web_ui.stop()

它有以下属性和方法：

user_classes：一个包含了所有用户类的列表，可以在运行时动态地添加或删除用户类。

shape_class：一个可选的Shape类，用于控制负载曲线。

tags：一个包含了所有标签的集合，用于过滤用户类或任务。

locustfile：一个包含了locustfile路径和模块对象的元组。

exclude_tags：一个包含了要排除的标签的集合。

events：一个EventHook对象，用于注册和触发事件钩子。

host：一个字符串，表示要测试的目标主机地址。

reset_stats：一个布尔值，表示是否在开始测试时重置统计数据。

stop_timeout：一个可选的整数，表示在停止测试时等待用户完成任务的最大秒数。

catch_exceptions：一个布尔值，表示是否捕获用户内部发生的异常并在UI/控制台中报告。

parsed_options：一个命名元组，包含了命令行参数的值。

create_local_runner()：为这个环境创建一个LocalRunner实例，用于单机模式下运行测试。

create_master_runner(master_bind_host, master_bind_port)：为这个环境创建一个MasterRunner实例，用于分布式模式下作为主节点运行测试。

create_web_ui(web_host, web_port, web_auth, tls_cert, tls_key)：为这个环境创建一个WebUI实例，并开始运行web服务器，用于提供web界面和RESTful API。

create_worker_runner(master_host, master_port)：为这个环境创建一个WorkerRunner实例，用于分布式模式下作为工作节点运行测试。

assign_equal_weights()：更新用户类，使每个用户以相同的概率运行其指定的任务。

locust自定义接口重构

from locust.env import Environment

env = Environment(user_classes=[MyHttpUser], events=events)
runner = env.create_local_runner()
# start a WebUI instance
web_ui = env.create_web_ui("127.0.0.1", 8089)
app=web_ui.app   #获取locust的app
@app.before_request
def stops():
    if request.path == "/stop":   #判断如果是stop接口走下面的逻辑
        runner.stop()
        res = get_html_report(env)
        with open('report.html', 'w') as f:
            f.write(res)
        return jsonify({"success": True, "message": "stopped"})

app=web_ui.app 获取到后，可以直接使用flask的方法就可以添加接口

修改locustweb界面，添加终止项目的按钮(locust 2.32.4 失效)

注册自定义的模板路径，模板位置在当前路径下的templates文件夹中

index_templates = Blueprint('index', __name__, template_folder='templates')
app.register_blueprint(index_templates)

修改请求路由的响应结果

@app.before_request
def before():
    if request.path == "/":
        return render_template("indexs.html", **web_ui.template_args)
		if request.path == "/stopWeb":
        web_ui.stop()
        runner.stop()
        runner.quit() #终止服务
        return jsonify({"success": True, "message": "stopped"})

HTML部分修改，在原来的HTML的基础上新增按钮，点按钮发起请求

<div >
    <a href="#" onclick='fetch("./stopWeb");' class="stop-button" style=" height: 27px; font-size: 8px; "><i></i>STOP WEB</a>
</div>

实现点击事件发送请求：onclick='fetch("./stopWeb");'

有关使用上的问题，欢迎您在底部评论区留言，一起交流~

[child_page]