Python軽量パフォーマンスツールLocustのインストールと使い方

Locust は Python のコルーチン メカニズムに基づいており、スレッド プロセスの制限を打ち破り、テスト マシン上で高い同時実行性を実現できます。

パフォーマンス テストの基本

1. 速度: パフォーマンスを測定します。システムの処理効率: 応答時間

2. どのくらい: システムの処理能力を測定します: 単位時間あたりに処理できるトランザクション数 (tps)

#パフォーマンス テストに基づいていますテスト要件の最も一般的な点について 次の 3 つのカテゴリについて

# 1 負荷テスト 負荷テスト

# 所定の指標に値するか、一部のシステム リソースがボトルネックに達すると、サーバーに継続的に負荷がかかります。目的は、システムの最大負荷容量を見つけることです

2 ストレス テスト

長時間続く高負荷でもシステムが安定しているかどうかを確認します

## 3 同時実行テスト:

同時にサーバーにリクエストを送信すると、システム内でトランザクションの競合やロックエスカレーションが発生しているかどうかが目的が判明します

パフォーマンス負荷モデル

locust インストール

インストールは存在します。質問です。Douban ソースから locust テンプレートをダウンロードできます。

pip install locust

locust テンプレート

基本的にほとんどのシナリオでは、このテンプレート read.py

from locust import HttpUser, TaskSet, task, tag, events
# 启动locust时运行
@events.test_start.add_listener
def setup(environment, **kwargs):
    # print("task setup")
# 停止locust时运行
@events.test_stop.add_listener
def teardown(environment, **kwargs):
    print("task teardown")
class UserBehavor(TaskSet):
    #虚拟用户启用task运行
    def on_start(self):
        print("start")
        locusts_spawned.wait()
    #虚拟用户结束task运行
    def on_stop(self):
        print("stop")
    @tag('test1')
    @task(2)
    def index(self):
        self.client.get('/yetangjian/p/17320268.html')
    @task(1)
    def info(self):
        self.client.get("/yetangjian/p/17253215.html")
class WebsiteUser(HttpUser):
    def setup(self):
        print("locust setup")
    def teardown(self):
        print("locust teardown")
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    min_wait = 3000
    max_wait = 5000

に基づいて変更を加えることができます。注: ここでは、ブラウザーで直接 locust を開くことができるように、ウェブホストが指定されています。

ランデブー lr_rendezvous

もちろん、ランデブー操作を上記のテンプレートのセットアップに組み込んで実行することもできます。

locusts_spawned = Semaphore()
locusts_spawned.acquire()
def on_hatch_complete(**kwargs):
    """
    select_task类的钩子函数
    :param kwargs:
    :return:
    """
    locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
n = 0
class UserBehavor(TaskSet):
    def login(self):
        global n
        n += 1
        print(f"第{n}个用户登陆")
    def on_start(self):
        self.login()
        locusts_spawned.wait()
    @task
    def test1(self):
        #catch_response获取返回
        with self.client.get("/yetangjian/p/17253215.html",catch_response=True):
            print("查询结束")
class WebsiteUser(HttpUser):
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    wait_time = between(1,3)
if __name__ == '__main__':
    os.system('locust -f read.py --web-host="127.0.0.1"')

一般的な使用法

上記の 2 つの例でいくつか見てきました。デコレータ events.test_start.add_listener; events.test_stop.add_listener は、負荷テストの前後にいくつかの操作を実行するために使用されます。別の例は、タスクの実行の前後に実行される on_start、on_stop です。別の例は、次の目的で使用できる task です。タスクの重みを割り当てます。

待機時間

# wait between 3.0 and 10.5 seconds after each task
#wait_time = between(3.0, 10.5)
#固定时间等待
# wait_time = constant(3)
#确保每秒运行多少次
constant_throughput（task_runs_per_second）
#确保每多少秒运行一次
constant_pacing(wait_time)

はユーザー クラスでも使用でき、wait_time を書き換えてカスタム

## を実現します。 #tag tag

@tag('test1')
@task(2)
def index(self):
    self.client.get('/yetangjian/p/17320268.html')

タスクをマークすると、実行時に特定のタスクを実行できます:

#只执行标记test1
os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"')
#不执行标记过的
os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"')
#除去test1执行所有
os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')

カスタマイズに失敗しました

#定义响应时间超过0.1就为失败
with self.client.get("/yetangjian/p/17253215.html", catch_response=True) as response:
    if response.elapsed.total_seconds() > 0.1:
        response.failure("Request took too long")
#定义响应码是200就为失败
with self.client.get("/yetangjian/p/17320268.html", catch_response=True) as response:
    if response.status_code == 200:
        response.failure("响应码200，但我定义为失败")

## カスタム ロード シェイプ

シェイプをカスタマイズします。py は LoadTestShape を継承し、tick

をオーバーライドします。このシェイプ クラスはユーザー数を 100 単位で増加します。 20 の割合でブロックし、10 分後に負荷テストを停止します (user_count は実行後の 51 秒目から 100 まで増加します)

from locust import LoadTestShape
class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20
    def tick(self):
        run_time = self.get_run_time()
        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, self.spawn_rate)
        return None

動作チャートは次のとおりです

#コマンド ラインを介したトリガー

os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')

#さまざまな時間段階でのトリガー例

from locust import LoadTestShape
class StagesShapeWithCustomUsers(LoadTestShape):
    stages = [
        {"duration": 10, "users": 10, "spawn_rate": 10},
        {"duration": 30, "users": 50, "spawn_rate": 10},
        {"duration": 60, "users": 100, "spawn_rate": 10},
        {"duration": 120, "users": 100, "spawn_rate": 10}]
    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data
        return None

以上がPython軽量パフォーマンスツールLocustのインストールと使い方の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。