在將locust作為第三方庫使用前，先對其中涉及的對象進(jìn)行介紹怕轿。
在了解關(guān)系下偷崩，便可更好的使用

一、locust架構(gòu)和類關(guān)系

介紹locust中涉及的對象撞羽，以及他們之間的關(guān)系
參考鏈接 https://testerhome.com/topics/24300
locust官方參考文檔 https://docs.locust.io/en/stable/

1.1. locust架構(gòu)

核心架構(gòu).png

• locust架構(gòu)上使用master-slave模型阐斜，支持單機(jī)和分布式
• master和slave（即worker）使用 ZeroMQ 協(xié)議通訊
• 提供web頁面管理master，從而控制slave诀紊，同時展示壓測過程和匯總結(jié)果
• 可選no-web模式（headless 一般用于調(diào)試）
• 基于Python本身已經(jīng)支持跨平臺

1.2. 主要類關(guān)系

先來一個關(guān)系圖谒出，看到locust主要類之間的關(guān)系

關(guān)系圖.png

簡單來說，Locust的代碼分為以下模塊：

• User-壓測用例：提供了HttpUser壓測http協(xié)議邻奠，用戶可以定義事務(wù)笤喳，斷言等，也可以實現(xiàn)特定協(xié)議的User
• Runner-執(zhí)行器：Locust的核心類碌宴，定義了整個框架的執(zhí)行邏輯杀狡，實現(xiàn)了Master、Slave（worker）等執(zhí)行器
• EventHook-事件鉤子：通過預(yù)先定義的事件使得我們可以在這些事件發(fā)生時(比如slave上報)做一些額外的操作
• WebU：提供web界面的操作臺和壓測過程展示
• Socket-通信器：提供了分布式模式下master和slave的交互方式
• RequestStats-采集贰镣、分析器：定義了結(jié)果分析和數(shù)據(jù)上報格式

1.3 核心類

核心類圖.png

• 用戶定義的User類作為Runner的user_classes傳入
• TaskSet和User持有client呜象，可以在類中直接發(fā)起客戶端請求，client可以自己實現(xiàn)碑隆，Locust只實現(xiàn)了HttpUser
• master的client_listener監(jiān)聽施壓端client消息
• slave的worker方法監(jiān)聽master消息
• slave的stats_reporter方法上報壓測數(shù)據(jù)恭陡，默認(rèn)3s上報一次
• slave的start啟動協(xié)程，使用master分配的并發(fā)數(shù)開始壓測
• slave默認(rèn)1s上報一次心跳上煤，如果master超過3s未收到某個slave的心跳則會將其標(biāo)記為missing狀態(tài)

主要結(jié)構(gòu)介紹完了休玩，接下來看下具體的類和對應(yīng)的方法

二、用戶行為User task TaskSet

2.1. User

一個User代表一個壓測用戶。locust將為每個正在模擬的用戶生成User類的一個實例哥捕。
【User可定義公共屬性】

• 2.1.1. wait_time屬性：單位秒，兩次執(zhí)行task時間的間隔嘉熊。between遥赚、constant、constant_pacing
  eg：自定義wait_time下面的User類將休眠一秒鐘阐肤，然后休眠兩個凫佛，然后休眠三個，依此類推

class MyUser(User):
    last_wait_time = 0

    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

    ...

• 2.1.2. weight屬性：通過設(shè)置weight參數(shù)孕惜，設(shè)置用戶比例
  eg：網(wǎng)絡(luò)用戶的可能性是移動用戶的三倍

class WebUser(User):
    weight = 3
    ...

class MobileUser(User):
    weight = 1
    ...

如果文件中存在多個用戶類愧薛，并且在命令行上未指定任何用戶類，則Locust將產(chǎn)生相等數(shù)量的每個用戶類
可以通過將它們作為命令行參數(shù)傳遞衫画，來指定要從同一locustfile中使用哪些用戶類

locust -f locust_file.py WebUser MobileUser

• 2.1.3. host屬性：要加載的主機(jī)的URL前綴（即“ http://google.com ”）
  如果在用戶類中聲明了主機(jī)屬性毫炉，則--host 在命令行或Web請求中未指定no的情況下將使用該屬性。
  可以在命令行削罩、Web UI中修改該屬性瞄勾。
  優(yōu)先級： Web UI > 命令行（--host) > 代碼

• 2.1.4. task屬性： @task
  詳細(xì)內(nèi)容見接下來的：2.2task

• 2.1.5. 環(huán)境屬性： environment
  用戶正在其中運(yùn)行的引用
  與環(huán)境或runner其所包含的環(huán)境進(jìn)行交互

self.environment.runner.quit()

如果在獨立蝗蟲實例上運(yùn)行，則將停止整個運(yùn)行弥激。如果在工作程序節(jié)點上運(yùn)行进陡，它將停止該特定節(jié)點

2.2 task

• 2.2.1. 宣告任務(wù):
  為用戶類（或TaskSet）聲明任務(wù)以使用task裝飾器的典型方式。
  @task采用可選的weight參數(shù)微服，該參數(shù)可用于指定任務(wù)的執(zhí)行率
  eg：task2被選擇為task1的機(jī)會是兩倍

from locust import User, task, between

class MyUser(User):
    wait_time = between(5, 15)

    @task(3)
    def task1(self):
        pass

    @task(6)
    def task2(self):
        pass

• 2.2.2. 任務(wù)屬性：
  不常用

• 2.2.3. 標(biāo)記任務(wù)：
  通過使用標(biāo)記<locust.tag>裝飾器標(biāo)記任務(wù)

from locust import User, constant, task, tag

class MyUser(User):
    wait_time = constant(1)

    @tag('tag1')
    @task
    def task1(self):
        pass

    @tag('tag1', 'tag2')
    @task
    def task2(self):
        pass

    @tag('tag3')
    @task
    def task3(self):
        pass

    @task
    def task4(self):
        pass

2.3 TaskSet

用于模擬現(xiàn)實用戶分級操作的場景趾疚，單接口直接用User就可以。

TaskSet是蝗蟲任務(wù)的集合以蕴，將像直接在User類上聲明的任務(wù)一樣執(zhí)行糙麦，使用戶在兩次任務(wù)執(zhí)行之間處于休眠狀態(tài).

• 帶有TaskSet的Demo：

class ForumSection(TaskSet):
    @task(10)
    def view_thread(self):
        pass

    @task(1)
    def create_thread(self):
        pass

    @task(1)
    def stop(self):
        self.interrupt()

class LoggedInUser(User):
    wait_time = between(5, 120)
    tasks = {ForumSection:2}

    @task
    def index_page(self):
        pass

• 使用@task裝飾器：
  直接在User / TaskSet類下內(nèi)聯(lián)TaskSet

class MyUser(User):
    @task(1)
    class MyTaskSet(TaskSet):
        ...

• 多層嵌套：
  TaskSet類的任務(wù)可以是其他TaskSet類，從而可以將它們嵌套任何數(shù)量的級別丛肮。
  例如喳资，我們可以使用以下結(jié)構(gòu)定義TaskSet：

- Main user behaviour
  - Index page
  - Forum page
    - Read thread
      - Reply
    - New thread
    - View next page
  - Browse categories
    - Watch movie
    - Filter movies
  - About page

當(dāng)正在運(yùn)行的用戶線程選擇TaskSet類執(zhí)行時，將創(chuàng)建該類的實例腾供，然后執(zhí)行將進(jìn)入該TaskSet仆邓。
然后發(fā)生的事情是，將拾取并執(zhí)行TaskSet的任務(wù)之一
然后線程將進(jìn)入用戶的wait_time函數(shù)指定的持續(xù)時間（除非wait_time直接在TaskSet類上聲明了該函數(shù)伴鳖，在這種情況下节值，它將使用該函數(shù)）
然后從TaskSet的任務(wù)中選擇一個新任務(wù)，然后再次等待榜聂，依此類推搞疗。

2.4 User和TaskSet的關(guān)系

• 在執(zhí)行時傳遞的參數(shù)是對TaskSet實例的引用，而不是User實例
• 可以從TaskSet實例中通過訪問User實例
• TaskSets還包含一個便捷 client 屬性须肆，該屬性引用User實例上的client屬性
• TaskSet實例的屬性user指向其User實例匿乃，parent指向其父TaskSet實例
• 標(biāo)記TaskSet會將標(biāo)記自動應(yīng)用于所有TaskSet的任務(wù)
• 如果您在嵌套的TaskSet中標(biāo)記任務(wù)桩皿，那么即使未標(biāo)記TaskSet，蝗蟲也將執(zhí)行該任務(wù)

至此幢炸，已經(jīng)了解了locust主要類之間的關(guān)系泄隔，以及主要類的功能。
接下來將以第三方庫的方式宛徊，將locust引入到項目工程中

三佛嬉、 以庫的方式引入locust

3.1. 創(chuàng)建一個 Environment 實例

from locust.env import Environment

env = Environment(user_classes=[MyTestUser])

3.2. 創(chuàng)建 create_master_runner 或 create_worker_runner啟動Runner

env.create_local_runner()
env.runner.start(5000, hatch_rate=20)
env.runner.greenlet.join()

3.3. start a Web UI

env.create_local_runner()
env.create_web_ui()
env.web_ui.greenlet.join()

3.4. 完整Demo

import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer
from locust.log import setup_logging

setup_logging("INFO", None)


class User(HttpUser):
    wait_time = between(1, 3)
    host = "https://docs.locust.io"

    @task
    def my_task(self):
        self.client.get("/")

    @task
    def task_404(self):
        self.client.get("/non-existing-path")

# setup Environment and Runner
env = Environment(user_classes=[User])
env.create_local_runner()

# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)

# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))

# start the test
env.runner.start(1, hatch_rate=10)

# in 60 seconds stop the runner
gevent.spawn_later(60, lambda: env.runner.quit())

# wait for the greenlets
env.runner.greenlet.join()

# stop the web server for good measures
env.web_ui.stop()

3.5. 判斷當(dāng)前狀態(tài)闸天，退出locust壓測

3.5.1 增加監(jiān)聽的代碼

@events.quitting.add_listener
def results(environment, **kw):
    logging.error("------------bonnie--------------")
    if environment.stats.total.fail_ratio > 0.01:
        logging.error("Test failed due to failure ratio > 1%")
        environment.process_exit_code = 1
    elif environment.stats.total.avg_response_time > 10:
        logging.error("Test failed due to average response time ratio > 200 ms")
        environment.process_exit_code = 1
    elif environment.stats.total.get_response_time_percentile(0.95) > 300:
        logging.error("Test failed due to 95th percentile response time > 800 ms")
        environment.process_exit_code = 1
    else:
        environment.process_exit_code = 0

3.5.2. 修改監(jiān)聽暖呕，注冊到init上

只用在init上被注冊，在實際執(zhí)行時才能被調(diào)用

from locust import events
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner

def checker(environment):
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(1)
        # if environment.runner.stats.total.fail_ratio > 0.2:
        # print(f"fail ratio was {environment.runner.stats.total.fail_ratio}, quitting")
        if environment.stats.total.avg_response_time > 40:

            print(f"fail ratio was {environment.stats.total.avg_response_time}, quitting")
            environment.runner.quit()
            return


@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    # only run this on master & standalone
    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(checker, environment)

# 需要在創(chuàng)建完env之后進(jìn)行調(diào)用苞氮，才能起作用
on_locust_init(env)

四湾揽、一個完整的Demo

涉及到其他文件，需要加載才能正常運(yùn)行

from locust import events
from locust.env import Environment
from locust.stats import stats_printer
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner
from locust.log import setup_logging, logging

import gevent
import time
from config import ConfigStopCondition, ConfigLoadInfo

from flask import request, Response
from locust import stats as locust_stats, runners as locust_runners
from locust import events
from prometheus_client import Metric, REGISTRY, exposition

is_quitting = False


def checker(environment):
    global is_quitting
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(1)
        if environment.runner.stats.total.fail_ratio > ConfigStopCondition.fail_ratio:
            logging.error(f"Test failed due to failure ratio > {ConfigStopCondition.fail_ratio}, quitting")
            print(f"Test failed due to failure ratio > {ConfigStopCondition.fail_ratio}, quitting")
            is_quitting = True
        elif environment.stats.total.avg_response_time > ConfigStopCondition.avg_response_time:
            logging.error(f"Test failed due to average response time ratio > {ConfigStopCondition.avg_response_time}, quitting")
            print(f"Test failed due to average response time ratio > {ConfigStopCondition.avg_response_time},quitting")
            is_quitting = True
        elif environment.stats.total.get_response_time_percentile(0.95) > ConfigStopCondition.response_time_95:
            logging.error(f"Test failed due to 95th percentile response time > {ConfigStopCondition.response_time_95}, ms quitting")
            print(f"Test failed due to 95th percentile response time > {ConfigStopCondition.response_time_95}, ms quitting")
            is_quitting = True

        if is_quitting:
            logging.error("Fail Ratio \t | Avg time \t | 95 time")
            logging.error(f" {environment.runner.stats.total.fail_ratio}  \t | "
                          f"{environment.stats.total.avg_response_time}  \t |  "
                          f"{environment.stats.total.get_response_time_percentile(0.95)} ")
            environment.runner.quit()
            return

@events.init.add_listener
def on_locust_init(environment, runner, **_kwargs):
    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(checker, environment)

def run_load_test(my_user):
    global is_quitting
    # 通過for循環(huán)笼吟，實現(xiàn)分不同用戶數(shù)量的壓測
    for u, r, rtime in zip(ConfigLoadInfo.user_list, ConfigLoadInfo.rate_list, ConfigLoadInfo.runtime_list):
        if not is_quitting:
            print( f"Current user is {u}")
            logging.error(f"Current user is {u}")
            # setup Environment and Runner
            env = Environment(user_classes=[my_user], step_load=True, stop_timeout=rtime*60*2)

            env.create_local_runner()

            # start a WebUI instance
            env.create_web_ui("**.**.**.**", 8089)

            # start a greenlet that periodically outputs the current stats
            gevent.spawn(stats_printer(env.stats))

            on_locust_init(env, env.runner)

            # start the test
            env.runner.start(u, hatch_rate=r)

            # in 60 seconds stop the runner
            gevent.spawn_later(rtime*60, lambda: env.runner.quit())

            # wait for the greenlets
            env.runner.greenlet.join()

            # stop the web server for good measures
            env.web_ui.stop()

if __name__ == "__main__":
    run_load_test(MyUser)

此時钝腺，已經(jīng)實現(xiàn)將locust作為第三方庫在python工程中運(yùn)行了
此時可以打開locust頁面，查看運(yùn)行狀態(tài)

通過上述代碼赞厕。實現(xiàn)了艳狐，以第三方庫的形式，分階段壓測被測對象皿桑。
并在不滿足判定條件時毫目，結(jié)束壓測。

下一小結(jié)诲侮，介紹通過Prometheus和Garapha對數(shù)據(jù)進(jìn)行長久保存镀虐。
并且生成可視化圖表