Locustとは？Pythonで書く負荷試験の使い方・ramp-up・分散実行【2.46対応】

Locust（ローカスト）は、負荷試験のシナリオをPythonのコードで書くオープンソースのツールです。この記事では2026年9月時点の最新版 Locust 2.46.5 を手元で動かし、ログイン付きAPIへの負荷試験を例に、シナリオの書き方からヘッドレス実行、段階的な負荷のかけ方、CIでの合否判定、分散実行までを実行結果つきで解説します。

まとめ：Locustで負荷試験を始める要点

• Locustは負荷試験用のPythonライブラリ兼コマンドです。最新版は2.46.5（2026年9月7日公開）で、Python 3.11以上が必要です。ライセンスはMITです。
• 導入は pip install locust、実行は locust -f locustfile.py です。Web UIは既定で8089番ポートに立ち上がります。
• CIでは --headless -u -r -t で条件を固定し、--csv と --html で結果を残します。失敗リクエストが1件でもあると終了コードは1になります。
• 秒間リクエスト数（RPS）は直接指定せず、ユーザー数と1サイクルの時間（待ち時間＋応答時間）から決まります。ユーザーごとのタスク実行回数を秒間の目標値に合わせたいときは constant_throughput を使います。
• 1プロセスで使えるCPUは1コアまでです。負荷が足りないときは、まず FastHttpUser に切り替え、それでも足りなければmaster/worker構成でコア数だけworkerを増やします。

Locustの特徴とJMeter・k6との違い

Pythonのコードでシナリオを書く設計

Locustでは、仮想ユーザーの振る舞いを User クラスとして書き、その中のメソッドを @task でタスクに登録します。GUIでテスト計画を組み立てるJMeterと違い、ログイン、トークンの付け替え、テストデータの読み込みといった処理を普通のPythonで書けます。シナリオがただのコードなので、Gitでの差分レビューもしやすくなります。

公式リポジトリは locustio/locust で、ライセンスはMITです。Python製のため、1プロセスは1コアしか使えません（GILによる制約）。高い負荷をかけるにはプロセスを増やす前提で設計されています。

JMeter・k6・Gatlingとの選び分け

チームの主な言語がPythonで、認証や前処理を含む込み入ったシナリオを書くならLocustが合います。逆に、テスト担当者がコードを書かない体制なら、Locustを第一候補にすべきではありません。また、Locustの1プロセスは1コアまでしか使えないため、高いRPSが必要な試験ではworkerを複数並べる構成が前提になります。JavaScriptでシナリオを書き、閾値で合否を判定する使い方は k6の記事 で、HTTP/2の性能試験は JMeterのHTTP/2プラグインの解説 で扱っています。

Locustのインストールと動作確認

pipで入れるのが基本です。WindowsでもmacOS・Linuxと同じ手順で入ります。システムのPythonを汚さないよう、仮想環境に入れるのが無難です。

python3 -m venv .venv
source .venv/bin/activate      # Windowsは .venv\Scripts\activate
pip install locust==2.46.5    # 最新版を入れるなら版指定を外す
locust --version
# locust 2.46.5 from .../site-packages/locust (Python 3.13.14)

Python 3.10以前ではLocust 2.46.5を入れられません。PyPIでの要件が >=3.11 になっているためです。Dockerを使う場合は公式イメージ locustio/locust に、カレントディレクトリのlocustfileをマウントして起動します。

docker run -p 8089:8089 -v $PWD:/mnt/locust locustio/locust -f /mnt/locust/locustfile.py

locustfileの書き方：ログイン付きAPIのシナリオ

HttpUserとタスクの重み

次のコードは、ログインしてトークンを受け取り、商品APIと遅いAPIを3対1の比率で呼ぶシナリオです。on_start は仮想ユーザーごとに1回だけ実行されるので、ログインはここに書きます。

import random
from locust import HttpUser, task, between


class ApiUser(HttpUser):
    wait_time = between(1, 2)

    def on_start(self):
        res = self.client.post("/api/login", json={"user": "demo", "password": "secret"})
        self.client.headers["Authorization"] = f"Bearer {res.json()['token']}"

    @task(3)
    def get_item(self):
        item_id = random.randint(1, 100)
        with self.client.get(f"/api/items/{item_id}", name="/api/items/[id]", catch_response=True) as res:
            if res.status_code == 404:
                res.success()

    @task(1)
    def slow(self):
        self.client.get("/api/slow")

@task(3) と @task(1) の数字は選ばれる確率の重みです。wait_time = between(1, 2) は、タスクを1回終えるごとに1〜2秒待つ指定です。self.client は requests のセッションと同じように扱え、Cookieやヘッダーはユーザーごとに保持されます。

URLの集約と、404を成功として数える判定

IDをURLに埋め込むと、統計がURLごとに別々の行へ分かれます。name="/api/items/[id]" を渡すと1行にまとまります。

存在しないIDへの404も、既定では失敗として数えられます。仕様どおりの応答なら、catch_response=True で応答を受け取り、res.success() を呼んで成功扱いにします。反対に、ステータスが200でも本文にエラーが入っている場合は res.failure("理由") で失敗にできます。

ヘッドレス実行とWeb UIの使い分け

コマンドラインの主要オプション

locust -f locustfile.py だけで起動すると、Web UI（http://localhost:8089）でユーザー数と生成レートを入力してから開始します。CIや再現性が必要な計測では、--headless で画面を使わずに条件を固定します。

locust -f locustfile.py -H http://127.0.0.1:8099 \
  --headless -u 20 -r 5 -t 20s \
  --csv out/run1 --html report.html --only-summary

コマンドラインの -t は、--headless か --autostart と組み合わせたときだけ有効です。Web UIから開始する場合は、開始フォームのRun time欄で実行時間を指定します。指定しなければ、止めるまで走り続けます。

Web UIのタブと見る場所

2.46.5のWeb UIには、Statistics、Charts、Failures、Exceptions、Current Ratio、Download Data、Logsのタブがあります。分散実行のときはWorkersタブも加わります。実行中に見るのは主に次の3つです。

• Statistics：リクエスト名ごとの件数、失敗数、中央値、95パーセンタイル、RPS
• Charts：RPS、応答時間、ユーザー数の推移。ユーザー数を増やしてもRPSが伸びず応答時間だけ上がり始めた点が、限界の目安です。ただし負荷生成側のCPUが足りない場合も同じ形になるので、Locustのコンソールに出るCPU使用率の警告も確認します
• Failures：失敗の種類と件数。ステータスコードや例外メッセージ単位で集計されます

Download DataからはCSVとHTMLレポートを取得できます。

RPSの決まり方とconstant_throughputによる調整

Locustで指定するのはユーザー数で、RPSは直接指定しません。1タスクで1リクエストを送るシナリオなら、RPSは「ユーザー数 ÷ 1サイクルの時間（待ち時間＋応答時間）」がおおよその目安です。1タスクで複数のリクエストを送る場合は、その本数を掛けて考えます。先ほどのシナリオを20ユーザー（1秒に5人ずつ増加）で20秒実行すると、次の集計になりました。

Type     Name              # reqs   # fails |  Avg   Min   Max   Med |  req/s
GET      /api/items/[id]      144   0(0.00%) |    6     1    39     5 |   7.40
POST     /api/login            20   0(0.00%) |    9     4    14     9 |   1.03
GET      /api/slow             69   0(0.00%) |  432   304   478   450 |   3.55
         Aggregated           233   0(0.00%) |  132     1   478     6 |  11.98

待ち時間の平均は1.5秒なので、20ユーザーがそろった状態の目安は約13 req/sです。実測の合計11.98 req/sには、20人に達するまでの約3秒間と、ログインの20件が含まれています。そのうえで、/api/slow の応答に約0.45秒かかるぶんサイクルが長くなり、その分だけRPSが下がります。つまり、対象システムが遅くなるとRPSも下がります。「ユーザー数を固定して応答時間を見る試験」ではこれが自然ですが、「毎秒20件を処理できるか」を確かめたい試験では条件がぶれます。

RPSを目標値に近づけたいときは、wait_time = constant_throughput(2) のように、1ユーザーが1秒間にタスクを何回実行するかを指定します。調整されるのはリクエスト数ではなくタスクの実行回数です。商品APIを呼ぶタスク1つだけのシナリオに変えて、10ユーザー・15秒で実行すると、商品APIは20.38 req/sになりました（ログインを含む合計は21.06 req/s）。ただし、これは待ち時間を詰めて目標に近づける仕組みです。タスクの実行に0.5秒以上かかると待ち時間は0になり、目標のRPSには届きません。

LoadTestShapeによるramp-upと段階的な負荷

-r は一定の割合でユーザーを増やすだけです。「5人で10秒、15人に増やして10秒、その後0人」のような段階的な負荷は、LoadTestShape を継承したクラスで書きます。Locustは tick() を約1秒ごとに呼び、戻り値の「ユーザー数と生成レート」に合わせて負荷を調整します。None を返すとテストが終わります。

from locust import HttpUser, LoadTestShape, task, constant


class WebUser(HttpUser):
    wait_time = constant(1)

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


class StepShape(LoadTestShape):
    # (終了秒, ユーザー数, 生成レート)
    stages = [(10, 5, 5), (20, 15, 5), (30, 0, 5)]

    def tick(self):
        run_time = self.get_run_time()
        for end, users, rate in self.stages:
            if run_time < end:
                return (users, rate)
        return None

locustfileにShapeクラスがあると、-u -r -t は不要で、接続先の -H と --headless だけで実行できます。実行ログでは、開始直後に5人、10秒後に15人、20秒後に0人へ切り替わり、30秒の時点で終了しました。

Shape test updating to 5 users at 5.00 spawn rate
Shape test updating to 15 users at 5.00 spawn rate
Shape test updating to 0 users at 5.00 spawn rate

結果レポートの読み方：CSV・HTML・JSON

--csv out/run1 を付けると、2.46.5では次の4ファイルが出力されました。--help の説明文には3ファイルと書かれていますが、実際には例外用のファイルも作られます。

run1_stats.csv の列は Request Count、Failure Count、Median Response Time、Requests/s に続き、50%から100%までのパーセンタイルが並びます。応答時間の単位はミリ秒です。性能要件がパーセンタイルで決まっているなら、95%や99%の列と照らし合わせます。平均は少数の遅いリクエストに引っ張られるためです。今回の集計（Aggregated）では、平均132msに対して中央値は6msでした。

1ファイルで共有したいなら --html report.html でグラフ付きのHTMLレポートを出力します。ファイル名の {u} {r} {t} は、ユーザー数、生成レート、実行時間に置き換わります。別のプログラムで集計する場合は、--json（標準出力）か --json-file で最終集計をJSONとして受け取れます。

CIでの合否判定と終了コードの制御

ヘッドレス実行では、失敗リクエストが1件でもあるとLocustは終了コード1で終わります。商品APIの404を失敗のまま数えるシナリオで試すと、失敗率10.69%で Shutting down (exit code 1) になりました。GitHub ActionsなどのCIでは、このステップがそのまま失敗します。

「失敗は1%まで許す」「95パーセンタイルが800msを超えたら落とす」といった条件にしたい場合は、quitting イベントで environment.process_exit_code を設定します。

import logging
from locust import events


@events.quitting.add_listener
def check_result(environment, **kwargs):
    stats = environment.stats.total
    if stats.num_requests == 0:
        logging.error("リクエストが1件も記録されていません")
        environment.process_exit_code = 1
    elif environment.runner.exceptions:
        logging.error("locustfile内で例外が発生しました")
        environment.process_exit_code = 1
    elif stats.fail_ratio > 0.01:
        logging.error("失敗率が1%を超えました")
        environment.process_exit_code = 1
    elif stats.get_response_time_percentile(0.95) > 800:
        logging.error("95パーセンタイルが800msを超えました")
        environment.process_exit_code = 1
    else:
        environment.process_exit_code = 0

process_exit_code に値を入れると、Locust標準の判定（失敗が1件でもあれば1）より優先されます。1%以下の失敗を許すには合格時に0を入れる必要がありますが、そのままではlocustfile内の例外も見逃します。そのため、この例では environment.runner.exceptions を先に確認し、リクエストが1件も記録されない場合（ユーザー数0など）も不合格にしています。

このコードで試すと、失敗率0.53%（1,500件中8件）では終了コード0、タスク内で例外を送出すると1、ユーザー数0でも1になりました。応答に約0.45秒かかるAPIだけを流して閾値を300msに下げた場合も1です。失敗があっても終了コードを変えたくないときは、--exit-code-on-error 0 を指定します（既定値は1）。性能要件を数値で決める手順は 非機能要件の数値化と閾値テストの解説 を参照してください。

分散実行：masterとworkerの構成

masterとworkerの起動手順

分散実行では、--master を付けたプロセスがWeb UIと全体の制御を担当し、--worker を付けたプロセスが仮想ユーザーを動かして統計をmasterへ送ります。master自身はユーザーを動かしません。1プロセスは1コアしか使えないため、負荷生成マシンのコア数と同じ数だけworkerを立てます。

# 負荷生成マシンA（master）
locust -f locustfile.py --master -H https://api.example.com \
  --headless -u 1000 -r 50 -t 10m --expect-workers 4

# 負荷生成マシンB・C（workerを2つずつ）
locust -f locustfile.py --worker --master-host 192.168.1.10

--expect-workers 4 は、workerが4つ接続するまで開始を待つ指定です。-u や -r はmaster側だけに付ければ、workerへ配分されます。masterとworkerは5557番ポートで通信するので、ファイアウォールで開けておきます。

1台のマシンでコア数分を使うだけなら、--processes で自動的にプロセスを分けられます。--processes 2 で実行したところ、workerが2つ接続されて負荷を分担しました。ただしこのオプションはWindowsでは使えません。この構成でも終了コードはmasterのものが返り、失敗があるときは1になりました。

Docker ComposeとKubernetes Operator

Docker Composeでは、公式イメージをmasterとworkerの2サービスとして定義します。公式ドキュメントの例では -H がmaster自身のWeb UI（http://master:8089）になっているので、試験対象のURLに書き換えて使います。

services:
  master:
    image: locustio/locust
    ports:
     - "8089:8089"
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --master -H https://api.example.com

  worker:
    image: locustio/locust
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --worker --master-host master

workerを増やすときは docker compose up --scale worker=4 とします。Kubernetesでは、公式ドキュメントに載っているLocust Operatorが使えます。master/workerのJob作成、locustfileのマウント、Web UIの公開、メトリクス収集までをカスタムリソースで管理できます。なお、マネージドサービスのLocust Cloudは終了に向かっており、2.43.0（2025年12月30日）で本体からの参照が削除されました。

FastHttpUserで負荷生成側の性能を上げる

workerを増やす前に試したいのが FastHttpUser です。HttpUser と同じAPIのまま、内部のHTTPクライアントを python-requests から geventhttpclient に切り替えます。公式ドキュメントでは、小さなリクエストを待ち時間なしで繰り返す最良のケースで、1プロセスあたり HttpUser が約4,000 req/s、FastHttpUser が約16,000 req/sとされています（2021年のM1 MacBook Pro、Python 3.11で計測）。

1ユーザー・待ち時間0で10秒ずつ比べた結果は、HttpUser が463.88 req/s、FastHttpUser が854.88 req/sで、約1.8倍でした（Intel Core i9-9880H、検証用サーバーはPython標準の ThreadingHTTPServer を同じマシンで稼働）。公式の4倍とは、マシン、Pythonの版、応答側サーバーの条件がいずれも異なるため、倍率は環境ごとに測り直す必要があります。負荷生成側のCPUが先に張り付くかどうかは、試験中に top などで確認します。

切り替えは from locust import FastHttpUser にして継承元を変えるだけです。ただし requests 固有の機能（セッションのアダプター設定など）に依存したコードは、そのままでは動きません。

HTTP以外のプロトコルとlocust-plugins

Locustの計測はHTTPに限りません。任意の処理の所要時間を events.request.fire で報告すれば、Statisticsに1行として並びます。次は、TCP接続にかかる時間を計測する例です。

import socket
import time
from locust import User, task, constant


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

    @task
    def connect(self):
        start = time.perf_counter()
        exc = None
        try:
            with socket.create_connection(("127.0.0.1", 8099), timeout=2):
                pass
        except OSError as e:
            exc = e
        self.environment.events.request.fire(
            request_type="TCP",
            name="connect",
            response_time=(time.perf_counter() - start) * 1000,
            response_length=0,
            exception=exc,
            context={},
        )

実行すると、集計に TCP connect の行が追加され、HTTPのリクエストと同じ形式で件数と応答時間が出ました。HTTP以外のプロトコルについて、公式に用意されているものは次のとおりです。

• gRPC：本体に同梱されたクラスではなく、公式ドキュメントとリポジトリの examples/grpc に、利用者が組み込む GrpcUser 基底クラスの例があります
• Socket.IO：本体同梱の SocketIOUser（実験的機能）
• MQTT：本体同梱の MqttUser（実験的機能）

Socket.IOではない素のWebSocketには、公式ドキュメントに専用クラスがありません。WebSocketクライアントのライブラリを上の例と同じ形で包み、送信から応答までの時間を events.request.fire で報告します。

コミュニティ製の拡張集 locust-plugins（SvenskaSpel、2026年6月に5.0.3）には、Playwright・Selenium・Kafka・FTP用のユーザー、CSVやMongoDBからテストデータを読むリーダー、--check-rps や --check-fail-ratio による合否判定オプションなどがあります。

よくある質問

Locustの読み方は？

「ローカスト」と読みます。英単語のlocustは「イナゴ」の意味で、大量の仮想ユーザーが一斉にシステムへ押し寄せる様子に由来します。

LocustはWindowsで使えますか？

使えます。pip install locust で導入でき、Web UI、ヘッドレス実行、master/workerの分散実行も動きます。1台で自動的にプロセスを分ける --processes だけはWindowsで使えないため、複数コアを使うときはworkerを個別に起動します。

Locustの負荷試験でramp-upを設定するには？

一定の割合で増やすだけなら、-r（1秒あたりに増やすユーザー数）を指定します。段階的に増減させるなら、LoadTestShape を継承したクラスの tick() で、経過時間に応じたユーザー数と生成レートを返します。

Locustは有料ですか？

本体は無料です。MITライセンスのオープンソースで、商用の負荷試験にも使えます。有料のマネージドサービスだったLocust Cloudは、2.43.0で本体からの参照が削除されています。

Locustとk6はどちらを選ぶべきですか？

シナリオをPythonで書きたいならLocust、JavaScriptやTypeScriptで書き、閾値による合否判定を標準機能で済ませたいならk6です。Locustで高いRPSが必要な場合は、workerを複数並べる前提になります。

関連記事

• k6とは？負荷テストツールの使い方・インストール手順を実例で解説
• 非機能要件とは？6大項目の数値化とk6での閾値テストまで実装目線で解説
• JMeterとHTTP/2プラグインの概要：HTTP/2性能テストの必要性と背景、導入メリットを解説
• Postmanとは？APIテスト自動化とNewmanでのCI組み込みまで実装手順で解説
• カオステストとは？障害注入の実験設計とツール選定・本番実施の判断を実装者向けに解説