首页 › Python论坛 › 【Python】万字长文，Locust 性能测试指北

到时候再说 發表於 2023-7-8 23:18:00

【Python】万字长文，Locust 性能测试指北

<h2 id="locust">Locust</h2>
<p>Locust 是比较常见的性能测试工具，底层基于 gevent。官方介绍 它是一款易于使用、可编写脚本且可扩展的性能测试工具，可以让我们使用常规 Python 代码定义用户的行为，而不必陷入 UI 或限制性领域特定语言中.</p>
<p><code>Locust</code>具有无限的可扩展性（<strong>只要提供客户端python 代码，适用于所有协议的性能测试</strong>）.</p>
<p>本文为开发性能自动化对比平台时学习相关内容的记录整理。</p>
<h2 id="我们为什么选择locust">我们为什么选择locust</h2>
<table>
<thead>
<tr>
<th>特点</th>
<th>说明</th>
</tr>
</thead>
<tbody>
<tr>
<td><strong>开源免费</strong></td>
<td>Locust是一个开源项目，无需支付费用，可以自由使用和定制。</td>
</tr>
<tr>
<td><strong>易于学习使用</strong></td>
<td>使用Python编写，学习路线平缓，拥有丰富的库和社区支持。</td>
</tr>
<tr>
<td><strong>可扩展性灵活性高</strong></td>
<td>可以根据需要定制测试，以便更准确地评估应用程序的性能。第三方插件较多、 易于扩展。</td>
</tr>
<tr>
<td><strong>实时统计</strong></td>
<td>提供实时统计功能和Web界面，方便监控和分析测试结果。</td>
</tr>
<tr>
<td><strong>易于集成</strong></td>
<td>可以轻松地与持续集成和持续部署工具集成，自动运行性能测试。</td>
</tr>
<tr>
<td><strong>适用大规模的性能测试</strong></td>
<td>支持分布式，可以轻松地在多台机器上运行测试，以模拟大量用户。这使得它非常适合进行大规模的性能测试。</td>
</tr>
</tbody>
</table>
<h2 id="locust的核心部件">Locust的核心部件</h2>
<h5 id="master节点">Master节点</h5>
<p>负责协调和管理整个测试过程，包括启动和停止测试、分发任务、收集和汇总测试结果等。</p>
<h5 id="worker节点">Worker节点</h5>
<p>实际执行测试任务的节点，根据Master节点分配的任务进行模拟用户行为。</p>
<h5 id="web-ui">Web UI</h5>
<p>提供可视化的测试界面，方便用户查看测试结果、监控测试进度等。</p>
<h5 id="测试脚本load-test-script">测试脚本（Load Test Script）</h5>
<p>测试脚本，定义模拟用户行为的逻辑和参数，由Worker节点执行。</p>
<h2 id="locust内部运行调用链路">Locust内部运行调用链路</h2>
<p>时序图如下：</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看时序图说明 </strong></summary>
<blockquote>
<ul>
<li>在测试启动时，Runner 类的 start() 方法会被调用，该方法会依次调用 EventHook 类的 fire() 方法，触发测试开始事件。</li>
<li>Runner 类会根据配置创建 Environment 类的实例，并将其作为参数传递给 User 类和 TaskSet 类的构造函数，同时将 User 类和 TaskSet 类添加到 Environment 类的 user_classes 属性中。</li>
<li>在测试运行期间，Runner 类会启动多个用户进程，每个用户进程都会创建一个 User 类的实例，并调用 User 类的 run() 方法，该方法会调用 TaskSet 类的 run() 方法，从而执行用户的任务。</li>
<li>在任务执行期间，User 类和 TaskSet 类会使用 Environment 类的 client 属性来发送请求，并使用 Environment 类的 stats 属性来记录统计信息。</li>
<li>在任务执行完成后，TaskSet 类的 run() 方法会返回，User 类的 run() 方法会进入等待状态，等待其他用户完成任务。</li>
<li>在测试结束时，Runner 类的 stop() 方法会被调用，该方法会依次调用 EventHook 类的 fire() 方法，触发测试结束事件。</li>
</ul>
</blockquote>
</details>
<div class="mermaid">sequenceDiagram
autonumber
Runner-&gt;&gt;EventHook: start()
EventHook-&gt;&gt;Environment: fire()
Runner-&gt;&gt;Environment: create Environment and add user classes
Runner-&gt;&gt;User: start users and run tasks
User-&gt;&gt;TaskSet: run()
TaskSet-&gt;&gt;Environment: send requests and record statistics
User-&gt;&gt;User: wait for other users to finish
TaskSet-&gt;&gt;User: return
Runner-&gt;&gt;User: stop users and tasks
Runner-&gt;&gt;EventHook: stop()
EventHook-&gt;&gt;Environment: fire()
</div><blockquote>
<p>注：fire() 方法是 Locust 中的 EventHook 类中的一个方法，用于触发事件。在 Locust 的测试生命周期中，有多个事件可以被触发，例如测试开始、测试结束、用户启动、用户完成任务等。当这些事件发生时，EventHook 类会调用 fire() 方法，将事件传递给所有注册了该事件的回调函数。</p>
</blockquote>
<h2 id="locust-实践">locust 实践</h2>
<h4 id="locust安装">locust安装</h4>
<table>
<thead>
<tr>
<th>步骤</th>
<th>说明</th>
</tr>
</thead>
<tbody>
<tr>
<td><strong>点击跳转安装 Python3.7+</strong></td>
<td>新版本标注需要Python3.7 or later</td>
</tr>
<tr>
<td><code>pip install locust </code></td>
<td>安装locust</td>
</tr>
<tr>
<td>执行<code>locust</code></td>
<td>检查是否安装成功</td>
</tr>
</tbody>
</table>
<pre><code>$ locust -V
locust 2.15.1 from /Users/bingohe/Hebinz/venvnew/lib/python3.9/site-packages/locust (python 3.9.17)
</code></pre>
<h4 id="入门示例">入门示例</h4>
<p>使用 locust 编写用例时，约定大于配置：</p>
<blockquote>
<p>test_xxx (一般测试框架约定)<br>
dockerfile (docker约定)<br>
locustfile.py (locust约定)</p>
</blockquote>
<pre><code># locustfile.py
from locust import HttpUser, task

class HelloWorldUser(HttpUser): # 父类是个User,表示要生成进行负载测试的系统的 HTTP“用户”。每个user相当于一个协程链接 ，进行相关系统交互操作
    @task# 装饰器来标记为一个测试任务, 表示用户要进行的操作:访问首页 → 登录 → 增、删改查
    def hello_world(self):
      wait_time = between(1, 5)
      # self.client发送 HTTP 请求，模拟用户的操作
      self.client.get("/helloworld")
</code></pre>
<h4 id="启动测试">启动测试</h4>
<h5 id="gui-模式启动-locust">GUI 模式启动 <code>locust</code></h5>
<p>在有<code>locustfile.py</code>文件的目录直接执行<code>locust</code>命令，然后访问：<code>http://0.0.0.0:8089/</code> 即可看到下面的界面：</p>
<pre><code>$ locust   
MacBook-Pro.local/INFO/locust.main: Starting web interface at http://0.0.0.0:8089 (accepting connections from all network interfaces)
MacBook-Pro.local/INFO/locust.main: Starting Locust 2.15.1
</code></pre>
<p><img src="https://img2023.cnblogs.com/blog/1172048/202307/1172048-20230708170444264-196223975.png" alt="image" loading="lazy"></p>
<p><strong>指标详解：</strong></p>
<ul>
<li>Number of users模拟用户数，默认 1</li>
<li>Spawn rate： 生产数 （每秒）、   =&gt;jmeter : Ramp-Up Period (in seconds)， 默认 1</li>
<li>Host (e.g. http://www.example.com)=&gt;测试目标 svr 的 绝对地址</li>
</ul>
<p>填写 host点击 start 之后就会对被测服务如<code>http://{host}/helloworld</code> 发起请求。请求统计数据如下：</p>
<p><img src="https://img2023.cnblogs.com/blog/1172048/202307/1172048-20230708171055681-2099936580.png" alt="image" loading="lazy"></p>
<p><strong>WebUI Tab说明：</strong></p>
<table>
<thead>
<tr>
<th>Tab名称</th>
<th>功能描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>New test</td>
<td>点击该按钮可对模拟的总虚拟用户数和每秒启动的虚拟用户数进行编辑</td>
</tr>
<tr>
<td>Statistics</td>
<td>类似于 JMeter 中 Listen 的聚合报告</td>
</tr>
<tr>
<td>Charts</td>
<td>测试结果变化趋势的曲线展示图，包括每秒完成的请求数（RPS）、响应时间、不同时间的虚拟用户数</td>
</tr>
<tr>
<td>Failures</td>
<td>失败请求的展示界面</td>
</tr>
<tr>
<td>Exceptions</td>
<td>异常请求的展示界面</td>
</tr>
<tr>
<td>Download Data</td>
<td>测试数据下载模块，提供三种类型的 CSV 格式的下载，分别是：Statistics、responsetime、exceptions</td>
</tr>
</tbody>
</table>
<p>需要说明的是webui 模式有很多限制，主要用于调试，下面将要介绍的命令行模式更为常用。</p>
<h5 id="命令行模式启动-locust">命令行模式启动 <code>locust</code></h5>
<pre><code>locust -f locustfile.py --headless -u 500 -r 10--host http://www.example.com-t 1000s
</code></pre>
<p>​ 框架是通过命令<code>locust</code>运行的，常用参数有：</p>
<table>
<thead>
<tr>
<th>参数</th>
<th>含义</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>-f</code> 或 <code>--locustfile</code></td>
<td>指定测试脚本文件的路径</td>
</tr>
<tr>
<td><code>--headless</code></td>
<td>以非 GUI 模式运行测试</td>
</tr>
<tr>
<td><code>-u</code> 或 <code>--users</code></td>
<td>指定并发用户数</td>
</tr>
<tr>
<td><code>-r</code> 或 <code>--spawn-rate</code></td>
<td>指定用户生成速率（即每秒生成的用户数）</td>
</tr>
<tr>
<td><code>-t</code> 或 <code>--run-time</code></td>
<td>指定测试运行的最大时间 （单位：秒），与--no-web一起使用</td>
</tr>
<tr>
<td><code>--csv</code></td>
<td>将测试结果输出到 CSV 文件中</td>
</tr>
<tr>
<td><code>--html</code></td>
<td>将测试结果输出为 HTML 报告</td>
</tr>
<tr>
<td><code>--host</code>或者 <code>-H</code></td>
<td>指定被测服务的地址</td>
</tr>
<tr>
<td><code>-L</code></td>
<td>日志级别，默认为INFO</td>
</tr>
</tbody>
</table>
<h2 id="检查点断言">检查点（断言）</h2>
<p>Locust默认情况下会根据HTTP状态码来判断请求是否成功。<strong>对于HTTP状态码范围在200-399之间的响应，Locust会将其视为成功。对于HTTP状态码在400-599之间的响应，Locust会将其视为失败。</strong></p>
<p>如果需要根据响应内容或其他条件来判断请求是否成功，需要手动设置检查点：</p>
<ul>
<li>使用self.client提供的catch_response=True`参数， 添加locust提供的ResponseContextManager类的上下文方法手动设置检查点。</li>
<li>ResponseContextManager里面的有两个方法来声明成功和失败，分别是<code>success</code>和<code>failure</code>。其中failure方法需要我们传入一个参数，内容就是失败的原因。</li>
</ul>
<pre><code class="language-python">from locust import HttpUser, task, between

class MyUser(HttpUser):
    # 思考时间：模拟真实用户在浏览应用程序时的行为
    wait_time = between(1, 5)

    @task
    def my_task(self):
      # 基于Locust提供的ResponseContextManager上下文管理器，使用catch_response=True 参数来捕获响应，手动标记成功或失败,
      with self.client.get("/some_page", catch_response=True) as response:
            # 检查状态码是否为200且响应中包含 "some_text"
            if response.status_code == 200 and "some_text" in response.text:
                # 如果满足条件，标记响应为成功
                response.success()
            else:
                # 如果条件不满足，根据具体情况生成错误信息
                error_message = "Unexpected status code: " + str(response.status_code) if response.status_code != 200 else "Expected text not found in the response"
                # 标记响应为失败，并报告错误信息
                response.failure(error_message)
</code></pre>
<h2 id="权重比例">权重比例</h2>
<p>如果需要请求有不同的比例，在Locust中，可以通过在<code>@task</code>装饰器中设置<code>weight</code>参数为任务分配权重来实现。权重越高，任务被执行的频率就越高。</p>
<pre><code class="language-python">from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(1, 5)

    # 设置权重为3，这个任务将被执行的频率更高
    @task(3)
    def high_frequency_task(self):
      self.client.get("/high_frequency_page")

    # 设置权重为1，这个任务将被执行的频率较低
    @task(1)
    def low_frequency_task(self):
      self.client.get("/low_frequency_page")
</code></pre>
<p>在这个示例中，我们为<code>high_frequency_task</code>任务设置了权重为3，而为<code>low_frequency_task</code>任务设置了权重为1。这意味着在模拟用户执行任务时，<code>high_frequency_task</code>任务被执行的频率将是<code>low_frequency_task</code>任务的3倍。通过设置权重，我们可以根据实际需求调整不同任务在性能测试中的执行频率。</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看在Locust内部权重的实现原理 </strong></summary>
<p>在Locust内部，权重是通过一个名为<code>TaskSet</code>的类来实现的。<code>TaskSet</code>类包含一个名为<code>tasks</code>的列表，该列表包含所有定义的任务。每个任务在列表中出现的次数等于其权重。当Locust选择要执行的任务时，它会从<code>tasks</code>列表中随机选择一个任务，这样权重较高的任务就有更高的概率被选中。</p>
<p>以下是一个简化的<code>TaskSet</code>类示例，以帮助理解权重是如何在Locust内部实现的：</p>
<pre><code class="language-python">import random

class TaskSet:
    def __init__(self):
      self.tasks = []

    def add_task(self, task, weight=1):
      for _ in range(weight):
            self.tasks.append(task)

    def get_random_task(self):
      return random.choice(self.tasks)

task_set = TaskSet()
task_set.add_task("high_frequency_task", weight=3)
task_set.add_task("low_frequency_task", weight=1)

# 当我们调用 get_random_task() 方法时，权重较高的任务有更高的概率被选中
random_task = task_set.get_random_task()
</code></pre>
<p>在这个示例中，我们创建了一个简化的<code>TaskSet</code>类，它包含一个<code>tasks</code>列表和两个方法：<code>add_task()</code>用于添加任务及其权重，<code>get_random_task()</code>用于随机选择一个任务。权重较高的任务在<code>tasks</code>列表中出现的次数更多，因此它们更有可能被<code>get_random_task()</code>方法选中。</p>
<p>在实际的Locust实现中，这个概念稍微复杂一些，但基本原理是相同的：通过权重调整任务在内部列表中出现的次数，从而影响任务被选中的概率。</p>
</details>
<h2 id="参数化">参数化</h2>
<p>在现实世界中，用户的行为通常是多样化的。他们可能使用不同的设备、操作系统、网络条件等。为了更好地模拟这些场景，我们需要在测试中使用不同的参数。</p>
<p>在性能测试中，参数化是一种非常重要的技术手段。它允许我们使用不同的数据集运行相同的测试场景，从而更好地模拟真实世界的用户行为。常用的参数化方法有两种。</p>
<h4 id="使用-locust-的内置参数化功能">使用 Locust 的内置参数化功能</h4>
<pre><code class="language-python">from locust import HttpUser, task, between
from locust.randoms import random_string, random_number

class MyUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def random_data(self):
      random_str = random_string(10)
      random_num = random_number(0, 100)
      self.client.post("/random", json={"text": random_str, "number": random_num})
</code></pre>
<h4 id="从外部文件读取参数">从外部文件读取参数</h4>
<p>以已经配置成白名单的鉴权 session 为例：</p>
<pre><code class="language-python">import csv
from locust import HttpUser, task


class CSVReader:
    def __init__(self, file, **kwargs):
      try:
            file = open(file)
      except TypeError:
            pass
      self.file = file
      self.reader = csv.reader(file, **kwargs)# iterator

    def __next__(self):
      try:
            return next(self.reader)
      except StopIteration:
            # 如果没有下一行，则从头开始读
            self.file.seek(0, 0)
            return next(self.reader)


session = CSVReader("session.csv")

class MyUser(HttpUser):
    @task
    def index(self):
      customer = next(ssn_reader)
      self.client.get(f"/pay?session={customer}")
</code></pre>
<h2 id="tag">Tag</h2>
<p>在<code>Locust</code>中，标签（Tag）是用于对任务进行分类和筛选的一种方法。通过给任务添加标签，可以在运行<code>Locust</code>时只执行具有特定标签的任务。这在执行特定场景的性能测试或组织大量任务时非常有用。</p>
<h5 id="使用场景">使用场景</h5>
<p>有时候我们会在同一个文件中写多个测试场景，但是运行的时候只想运行其中一部分，即当一个测试文件中的task不止一个时，我们可以通过<code>@tag</code>给task打标签进行分类，在执行测试时，通过<code>--tags name</code>执行指定带标签的task。</p>
<p>以下是一个使用标签的示例：</p>
<pre><code class="language-python">from locust import HttpUser, task, between, tag

class MyUser(HttpUser):
    wait_time = between(1, 5)

    # 给任务添加一个名为 "login" 的标签
    @tag("login")
    @task
    def login_task(self):
      self.client.post("/login", json={"username": "user", "password": "pass"})

    # 给任务添加一个名为 "profile" 的标签
    @tag("profile")
    @task
    def profile_task(self):
      self.client.get("/profile")

    # 给任务添加两个标签："shopping" 和 "checkout"
    @tag("shopping", "checkout")
    @task
    def checkout_task(self):
      self.client.post("/checkout")
</code></pre>
<p>在这个示例中，我们为三个任务分别添加了不同的标签。<code>login_task</code>任务具有<code>"login"</code>标签，<code>profile_task</code>任务具有<code>"profile"</code>标签，而<code>checkout_task</code>任务具有<code>"shopping"</code>和<code>"checkout"</code>两个标签。</p>
<p>运行Locust时，可以通过使用<code>--tags</code>和<code>--exclude-tags</code>选项来指定要执行或排除的标签。例如，要仅执行具有<code>"login"</code>标签的任务，可以运行：</p>
<pre><code class="language-bash">locust --tags login
</code></pre>
<p>要排除具有<code>"shopping"</code>标签的任务，可以运行：</p>
<pre><code class="language-bash">locust --exclude-tags shopping
</code></pre>
<p>这样，我们就可以根据需要执行特定场景的性能测试，而不需要修改代码。</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看如何在Locust属性中指定 tag </strong></summary>
<p>在Locust中，可以使用<code>tags</code>属性来在<code>HttpUser</code>子类中指定标签。以下是一个示例：</p>
<pre><code class="language-python">from locust import HttpUser, task, between, tag

@tag("login")
class LoginTasks(HttpUser):
    wait_time = between(1, 5)

    @task
    def login_task(self):
      self.client.post("/login", json={"username": "user", "password": "pass"})

@tag("profile")
class ProfileTasks(HttpUser):
    wait_time = between(1, 5)

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

@tag("shopping", "checkout")
class CheckoutTasks(HttpUser):
    wait_time = between(1, 5)

    @task
    def checkout_task(self):
      self.client.post("/checkout")
</code></pre>
<p>在这个示例中，我们创建了三个不同的<code>HttpUser</code>子类，分别为<code>LoginTasks</code>、<code>ProfileTasks</code>和<code>CheckoutTasks</code>。我们在类级别使用<code>@tag()</code>装饰器为每个子类添加了标签。<code>LoginTasks</code>具有<code>"login"</code>标签，<code>ProfileTasks</code>具有<code>"profile"</code>标签，而<code>CheckoutTasks</code>具有<code>"shopping"</code>和<code>"checkout"</code>两个标签。</p>
<p>与之前的示例类似，可以使用<code>--tags</code>和<code>--exclude-tags</code>选项来指定要执行或排除的标签。在这种情况下，标签将应用于整个<code>HttpUser</code>子类，而不仅仅是单个任务。</p>
</details>
<h2 id="集合点">集合点</h2>
<h4 id="什么是集合点">什么是集合点？</h4>
<p>集合点用以同步虚拟用户，以便恰好在同一时刻执行任务。在[测试计划]中，可能会要求系统能够承受1000 人同时提交数据，可以通过在提交数据操作前面加入集合点，这样当虚拟用户运行到提交数据的集合点时，就检查同时有多少用户运行到集合点，如果不到1000 人，已经到集合点的用户在此等待，当在集合点等待的用户达到1000 人时，1000 人同时去提交数据，从而达到测试计划中的需求。</p>
<blockquote>
<p>注意：Locust框架本身没有直接封装集合点的概念 ，需要间接通过gevent并发机制，使用gevent的锁来实现。</p>
</blockquote>
<p>在 Locust 中 实现集合点前，我们先了解两个概念：</p>
<ul>
<li><code>gevent</code> 中的 <code>Semaphore</code> 信号量</li>
<li><code>locust</code> 中的事件钩子 <code>all_locusts_spawned</code></li>
</ul>
<h5 id="semaphore">Semaphore</h5>
<p>信号量（Semaphore）是一种用于控制对共享资源访问的同步原语。它是计算机科学和并发编程中的一个重要概念，最早由著名计算机科学家Edsger Dijkstra于1960s提出。信号量用于解决多线程或多进程环境中的临界区问题，以防止对共享资源的竞争访问。</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看`Semaphore`实现原理 </strong></summary>
<p>信号量的工作原理是通过维护一个计数器来表示可用资源的数量。当一个线程或进程想要访问共享资源时，它需要请求信号量。信号量会检查其计数器值：</p>
<p>如果计数器值大于0，表示有可用资源，信号量会减少计数器值并允许线程或进程访问共享资源。<br>
如果计数器值等于0，表示没有可用资源，信号量会阻塞线程或进程，直到其他线程或进程释放资源。<br>
当线程或进程完成对共享资源的访问后，它需要释放信号量。此时，信号量会增加计数器值，表示资源已释放并可供其他线程或进程使用。</p>
<p>信号量通常有两种类型：</p>
<p>二进制信号量（Binary Semaphore）：计数器值只能为0或1。二进制信号量通常用于实现互斥锁（Mutex），以确保一次只有一个线程或进程访问共享资源。<br>
计数信号量（Counting Semaphore）：计数器值可以为任意正整数。计数信号量用于限制对共享资源的并发访问数量，以实现有限的资源池。<br>
许多编程语言和库都提供了信号量的实现，例如Python中的threading.Semaphore和gevent.lock.Semaphore。使用信号量可以帮助解决并发编程中的同步和资源竞争问题。</p>
<p><code>gevent.lock.Semaphore</code>是<code>gevent</code>库中提供的信号量实现。<code>gevent</code>是一个基于协程的Python并发库，使用轻量级的绿色线程（greenlet）提供高性能的并发。<code>gevent.lock.Semaphore</code>允许您在<code>gevent</code>协程中同步对共享资源的访问。</p>
<p>以下是<code>gevent.lock.Semaphore</code>的主要特点和使用方法：</p>
<ol>
<li>初始化：要创建一个信号量，您可以实例化<code>gevent.lock.Semaphore</code>类。在初始化时，可以选择设置信号量的初始值（默认值为1）。</li>
</ol>
<pre><code class="language-python">from gevent.lock import Semaphore

# 创建一个具有默认初始值（1）的信号量
sem = Semaphore()

# 创建一个具有自定义初始值（5）的信号量
sem_with_initial_value = Semaphore(value=5)
</code></pre>
<ol start="2">
<li>请求资源（acquire）：当协程需要访问共享资源时，它应该调用<code>Semaphore.acquire()</code>方法。如果信号量的计数器值大于0，<code>acquire()</code>方法将减少计数器值并立即返回。如果计数器值为0，<code>acquire()</code>方法将阻塞协程，直到其他协程释放资源。</li>
</ol>
<pre><code class="language-python">sem.acquire()
# 在此处访问共享资源
</code></pre>
<ol start="3">
<li>释放资源（release）：当协程完成对共享资源的访问后，它应该调用<code>Semaphore.release()</code>方法。这将增加信号量的计数器值，表示资源已释放并可供其他协程使用。</li>
</ol>
<pre><code class="language-python"># 完成对共享资源的访问
sem.release()
</code></pre>
<ol start="4">
<li>使用上下文管理器：<code>gevent.lock.Semaphore</code>还可以作为上下文管理器使用，以确保在访问共享资源的代码块结束时自动释放信号量。这可以简化代码并防止忘记释放信号量。</li>
</ol>
<pre><code class="language-python">with sem:
    # 在此处访问共享资源
# 信号量会在这里自动释放
</code></pre>
<p>总之，<code>gevent.lock.Semaphore</code>是<code>gevent</code>库中提供的信号量实现，用于在协程之间同步对共享资源的访问。通过使用<code>Semaphore.acquire()</code>和<code>Semaphore.release()</code>方法，您可以确保在<code>gevent</code>协程中正确处理并发访问。</p>
</details>
<h5 id="all_locusts_spawned-事件">all_locusts_spawned 事件</h5>
<p>在 Locust 中，事件是一个非常重要的概念。事件允许我们在 Locust 的生命周期中的特定时刻执行自定义的操作。通过监听和处理这些事件，我们可以扩展 Locust 的功能，以满足测试需求。</p>
<p><code>spawning_complete</code> 是 Locust 中的一个事件，表示所有的 Locust 用户（user）已经生成完成。当 Locust 开始运行测试并生成用户时，它会逐渐创建用户实例。一旦所有的用户都被创建，<code>spawning_complete</code> 事件就会被触发。你可以在这个事件中执行一些特定的操作，例如输出日志消息、收集统计信息或执行其他自定义操作。</p>
<p>要监听 <code>spawning_complete</code> 事件，你可以使用 <code>locust.events.spawning_complete</code> 事件钩子。例如：</p>
<pre><code class="language-python">from locust import events

@events.spawning_complete.add_listener
def on_spawning_complete():
    print("All users have been spawned!")
</code></pre>
<p>在这个示例中，当所有的用户生成完成时，我们会输出一条消息 "All users have been spawned!"。你可以根据需要替换为其他操作。</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看 `Locust` 生命周期中其他的事件 </strong></summary>
<div class="mermaid">sequenceDiagram
    participant S as Script
    participant L as Locust
    Note over S, L: Test Starts

    S-&gt;&gt;L: test_start
    L-&gt;&gt;S: on_test_start

    Note over S, L: Spawning Users
    S-&gt;&gt;L: spawning_start
    L-&gt;&gt;S: on_spawning_start

    loop for each user
      S-&gt;&gt;L: user_add
      L-&gt;&gt;S: on_user_add
    end

    S-&gt;&gt;L: spawning_complete
    L-&gt;&gt;S: on_spawning_complete

    Note over S, L: Running Test
    loop for each request
      S-&gt;&gt;L: request
      L-&gt;&gt;S: on_request
    end

    Note over S, L: Test Stops
    S-&gt;&gt;L: test_stop
    L-&gt;&gt;S: on_test_stop
</div><ol>
<li><code>test_start</code>：测试开始时触发。</li>
<li><code>spawning_start</code>：生成用户时触发。</li>
<li><code>user_add</code>：每个用户被添加时触发。</li>
<li><code>spawning_complete</code>：所有用户生成完成时触发。</li>
<li><code>request</code>：每个请求发生时触发。</li>
<li><code>test_stop</code>：测试停止时触发。</li>
</ol>
</details>
<p>了解完上面两个概念，接下来我们只需要两步走：</p>
<ul>
<li>在脚本启动时，使用all_locust_spawned.acquire() 阻塞进程</li>
<li>编写一个函数，在 用户全部创建完成时触发 all_locust_spawned.release()</li>
</ul>
<p>示例代码：</p>
<pre><code class="language-python">from locust import HttpUser, task, between
from gevent.lock import Semaphore
from locust import events

all_locust_spawned = Semaphore()
all_locust_spawned.acquire()# 阻塞


class MyUser(HttpUser):
    wait_time = between(1, 1)

    def on_start(self):
      global all_locust_spawned
      all_locust_spawned.wait(3)# 同步锁等待时间

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


# 添加集合点事件处理器
@events.spawning_complete.add_listener# 所有的Locust实例产生完成时触发
def on_spawning_complete(**_kwargs):
    global all_locust_spawned
    all_locust_spawned.release()
</code></pre>
<h2 id="分布式">分布式</h2>
<p>当我们需要大量的并发用户，而单个计算机可能无法生成足够的负载来模拟这种情况时，分布式压力测试可以解决这个问题，我们可以通过将压力测试分布到多个计算机上来生成更大的负载，并更准确地评估系统的性能。</p>
<h5 id="locust-的限制">Locust 的限制</h5>
<p>Locust 使用 Python 的 asyncio 库来实现异步 I/O，这意味着它可以充分利用多核 CPU 的性能。然而，由于 Python 的全局解释器锁（GIL）限制，单个 Python 进程无法充分利用多核 CPU。</p>
<p>为了解决这个问题，Locust 支持在单个计算机上运行多个从节点（worker node），这样可以充分利用多核 CPU 的性能。</p>
<p><strong>当在单台计算机上运行多个从节点时，每个从节点将运行在一个单独的进程中，从而避免了 GIL 的限制。这样，我们可以充分利用多核 CPU 的性能，生成更大的负载。</strong></p>
<h4 id="单机主从模式">单机主从模式</h4>
<blockquote>
<p>注意： slave 的节点数要小于等于本机的处理器数</p>
</blockquote>
<p>在单机主从模式下，主节点和从节点都运行在同一台计算机上。这种模式适用于在本地开发环境中进行压力测试，或者在具有多核 CPU 的单台服务器上进行压力测试。以下是在单机主从模式下实现分布式压力测试的步骤：</p>
<ol>
<li>
<p>安装 Locust：在计算机上安装 Locust，使用 <code>pip install locust</code> 命令进行安装。</p>
</li>
<li>
<p>编写 Locust 测试脚本：编写一个 Locust 测试脚本，这个脚本将在主节点和从节点上运行。将此脚本保存为 <code>locustfile.py</code>。</p>
</li>
<li>
<p>启动主节点：在计算机上运行 <code>locust --master</code> 命令启动主节点，监听默认端口（8089）。</p>
</li>
<li>
<p>启动从节点：在计算机上运行 <code>locust --worker --master-host 127.0.0.1</code> 命令启动一个从节点。根据需要，可以启动多个从节点。</p>
</li>
<li>
<p>运行分布式压力测试：访问 Locust 的 Web 界面（http://127.0.0.1:8089），开始测试。</p>
</li>
</ol>
<h5 id="单机模式下如何让每个从节点都运行在不同的-cpu-上">单机模式下，如何让每个从节点都运行在不同的 CPU 上</h5>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)"> 点击查看如何单机模式下，每个从节点都运行在不同的 CPU 上</strong></summary>
<p>在单机主从模式下，确保启动的多个从节点运行在不同的 CPU 核心上，可以通过为每个从节点设置 <code>taskset</code> 命令来实现。<code>taskset</code> 是一个 Linux 命令，可以用来设置进程的 CPU 亲和性，即将进程绑定到特定的 CPU 核心上运行。</p>
<p>以下是在单机主从模式下，确保启动的多个从节点运行在不同 CPU 核心上的步骤：</p>
<ol>
<li>
<p>启动主节点：在计算机上运行 <code>locust --master</code> 命令启动主节点，监听默认端口（8089）。</p>
</li>
<li>
<p>启动从节点：在计算机上运行以下命令启动从节点，并将其绑定到特定的 CPU 核心上：</p>
</li>
</ol>
<pre><code class="language-bash">taskset -c CORE_NUMBER locust --worker --master-host 127.0.0.1
</code></pre>
<p>其中，<code>CORE_NUMBER</code> 是要将从节点绑定到的 CPU 核心编号（从 0 开始）。例如，要将从节点绑定到第一个 CPU 核心上，可以运行以下命令：</p>
<pre><code class="language-bash">taskset -c 0 locust --worker --master-host 127.0.0.1
</code></pre>
<ol start="3">
<li>根据需要，可以启动多个从节点，并将它们分别绑定到不同的 CPU 核心上。例如，要将第二个从节点绑定到第二个 CPU 核心上，可以运行以下命令：</li>
</ol>
<pre><code class="language-bash">taskset -c 1 locust --worker --master-host 127.0.0.1
</code></pre>
<p>请注意，<code>taskset</code> 命令仅适用于 Linux 系统。在 Windows 或 macOS 上，可以尝试使用类似的工具，例如 Windows 上的 <code>start /affinity</code> 命令或 macOS 上的 <code>cpulimit</code> 工具。</p>
<p>通过使用 <code>taskset</code> 命令或类似的工具，我们可以确保在单机主从模式下，启动的多个从节点运行在不同的 CPU 核心上。这有助于充分利用多核 CPU 的性能，生成更大的负载。</p>
</details>
<h4 id="多机主从模式">多机主从模式</h4>
<p>操作与单机模式基本一样，访问 Locust 的 Web 界面时访问的时主节点的地址（http://MASTER_IP_ADDRESS:8089）。</p>
<p>因为主节点和从节点之间通过网络通信。因此，在选择主节点和从节点的计算机时，需要确保它们之间的网络连接畅通。<strong>此外，为了获得准确的测试结果，务必确保主节点和从节点之间的网络延迟较低。</strong></p>
<h4 id="分布式模式下的命令参数">分布式模式下的命令参数</h4>
<table>
<thead>
<tr>
<th>命令参数</th>
<th>说明</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>--master</code></td>
<td>将当前 Locust 实例作为主节点（master node）运行。</td>
</tr>
<tr>
<td><code>--worker</code></td>
<td>将当前 Locust 实例作为从节点（worker node）运行。</td>
</tr>
<tr>
<td><code>--master-host</code></td>
<td>指定主节点的 IP 地址或主机名。默认值为 <code>127.0.0.1</code>。</td>
</tr>
<tr>
<td><code>--master-port</code></td>
<td>指定主节点的端口号。默认值为 <code>5557</code>。</td>
</tr>
<tr>
<td><code>--master-bind-host</code></td>
<td>指定主节点绑定的 IP 地址或主机名。默认值为 <code>*</code>（所有接口）。</td>
</tr>
<tr>
<td><code>--master-bind-port</code></td>
<td>指定主节点绑定的端口号。默认值为 <code>5557</code>。</td>
</tr>
<tr>
<td><code>--expect-workers</code></td>
<td>指定主节点期望连接的从节点数量。默认值为 <code>1</code>。</td>
</tr>
</tbody>
</table>
<p><code>--expect-workers</code> 参数用于指定主节点期望连接的从节点数量。如果实际连接的从节点数量没有达到这个值，主节点会继续等待，直到足够的从节点连接上来。</p>
<p>在实际运行分布式压力测试时，主节点会在 Web 界面上显示连接的从节点数量。如果实际连接的从节点数量没有达到 <code>--expect-workers</code> 指定的值，你可以在 Web 界面上看到一个警告消息，提示你主节点正在等待更多从节点的连接。</p>
<h2 id="docker-运行locust">docker 运行locust</h2>
<p>使用 容器的方式运行 locust 的优势和缺点都非常明显：</p>
<table>
<thead>
<tr>
<th>优势</th>
<th>描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>环境一致性</td>
<td>Docker 可以确保在不同计算机上运行的 Locust 环境是一致的。</td>
</tr>
<tr>
<td>便于部署</td>
<td>使用 Docker 可以简化 Locust 的部署过程。</td>
</tr>
<tr>
<td>易于扩展</td>
<td>Docker 可以与容器编排工具结合使用，实现 Locust 从节点的自动扩展。</td>
</tr>
<tr>
<td>隔离性</td>
<td>Docker 容器提供了一定程度的隔离性，将 Locust 运行环境与宿主机系统隔离。</td>
</tr>
</tbody>
</table>
<table>
<thead>
<tr>
<th>缺点</th>
<th>描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>性能开销</td>
<td>Docker 容器可能存在一定程度的性能损失，与在宿主机上直接运行 Locust 相比。</td>
</tr>
<tr>
<td>学习曲线</td>
<td>对于不熟悉 Docker 的用户，可能需要一定时间学习 Docker 的基本概念和使用方法。</td>
</tr>
<tr>
<td>系统资源占用</td>
<td>运行 Docker 容器需要消耗一定的系统资源（如 CPU、内存、磁盘空间等）。</td>
</tr>
</tbody>
</table>
<p>但是以下这些场景使用 Docker 来运行 Locust 是一个更好的选择：</p>
<ol>
<li>
<p><strong>分布式压力测试</strong>：在分布式压力测试中，需要在多台计算机上运行 Locust 主节点和从节点。使用 Docker 可以确保所有节点的运行环境一致，简化部署过程。</p>
</li>
<li>
<p><strong>云环境部署</strong>：如果你需要在云环境（如 AWS、Azure、GCP 等）中进行压力测试，使用 Docker 可以简化部署过程，并充分利用云平台提供的容器服务（如 Amazon ECS、Google Kubernetes Engine 等）。</p>
</li>
<li>
<p><strong>CI/CD 集成</strong>：如果你需要将压力测试集成到持续集成/持续部署（CI/CD）流程中，使用 Docker 可以简化集成过程。许多 CI/CD 工具（如 Jenkins、GitLab CI、Travis CI 等）都支持 Docker 集成。</p>
</li>
<li>
<p><strong>避免环境冲突</strong>：如果你的开发或测试环境中已经安装了其他 Python 应用程序，可能会出现依赖项冲突。使用 Docker 可以将 Locust 运行环境与宿主机系统隔离，避免潜在的环境冲突。</p>
</li>
<li>
<p><strong>团队协作</strong>：在团队协作过程中，使用 Docker 可以确保每个团队成员都使用相同的 Locust 运行环境，从而避免因环境差异导致的问题。</p>
</li>
</ol>
<h4 id="具体使用步骤">具体使用步骤</h4>
<ol>
<li>
<p>首先，确保你已经安装了 Docker。如果尚未安装，请参考 Docker 官方文档 以获取适用于你的操作系统的安装说明。</p>
</li>
<li>
<p>编写一个 Locust 测试脚本。例如，创建一个名为 <code>locustfile.py</code> 的文件，内容如下：</p>
</li>
</ol>
<pre><code class="language-python">from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def my_task(self):
      self.client.get("/")
</code></pre>
<ol start="3">
<li>使用以下命令从 Docker Hub 拉取官方的 Locust 镜像：</li>
</ol>
<pre><code class="language-bash">docker pull locustio/locust
</code></pre>
<ol start="4">
<li>使用以下命令在 Docker 中运行 Locust。</li>
</ol>
<pre><code class="language-bash">docker run --rm -p 8089:8089 -v $PWD:/mnt/locust locustio/locust -f /mnt/locust/locustfile.py --host TARGET_HOST
</code></pre>
<p>在这个命令中，我们将当前目录（包含 <code>locustfile.py</code> 文件）挂载到 Docker 容器的 <code>/mnt/locust</code> 目录。然后，我们使用 <code>-f</code> 参数指定要运行的 Locust 测试脚本，并使用 <code>--host</code> 参数指定目标主机地址。</p>
<ol start="5">
<li>访问 Locust 的 Web 界面。在浏览器中打开 <code>http://localhost:8089</code>，你将看到 Locust 的 Web 界面。在这里，你可以开始压力测试并查看结果。</li>
</ol>
<p>通过以上步骤，你可以在 Docker 中运行 Locust，无需在本地环境中安装 Locust。</p>
<p>总之，在需要确保环境一致性、简化部署过程、集成到 CI/CD 流程、避免环境冲突或团队协作的场景下，使用 Docker 来运行 Locust 是一个很好的选择。通过使用 Docker，你可以轻松地在不同的计算机或云环境中运行压力测试，从而实现更大规模的分布式压力测试。</p>
<h2 id="高性能-fasthttpuser">高性能 FastHttpUser</h2>
<p>Locust 的默认 HTTP 客户端使用<code>http.client</code>。如果计划以非常高的吞吐量运行测试并且运行 Locust 的硬件有限，那么它有时效率不够。</p>
<p><code>FastHttpUser</code> 是 <code>Locust</code> 提供的一个特殊的用户类，用于执行 HTTP 请求。与默认的 HttpUser 不同，FastHttpUser 使用 C 语言库 gatling 编写的 httpclient 进行 HTTP 请求, 有时将给定硬件上每秒的最大请求数增加了 5 到 6 倍。<strong>在相同的并发条件下使用FastHttpUser能有效减少对负载机的资源消耗从而达到更大的http请求。</strong></p>
<h4 id="优势">优势</h4>
<ol>
<li>
<p><strong>性能</strong>：<code>FastHttpUser</code> 的主要优势是性能。由于它使用 C 语言库进行 HTTP 请求，它的性能通常比默认的 <code>HttpUser</code> 更高。这意味着在相同的硬件资源下，你可以使用 <code>FastHttpUser</code> 生成更大的负载。</p>
</li>
<li>
<p><strong>资源占用</strong>：与默认的 <code>HttpUser</code> 相比，<code>FastHttpUser</code> 通常具有较低的资源占用（如 CPU、内存等）。这意味着在进行压力测试时，你可以在同一台计算机上运行更多的并发用户。</p>
</li>
<li>
<p><strong>更高的并发能力</strong>：由于 <code>FastHttpUser</code> 的性能和资源占用优势，它可以更好地支持大量并发用户的压力测试。这对于需要模拟大量并发用户的场景（如高流量 Web 应用程序、API 等）非常有用。</p>
</li>
</ol>
<p><strong>然而需要注意的是<code>FastHttpUser</code> 也有一些局限性。例如，它可能不支持某些特定的 HTTP 功能（如自定义 SSL 证书、代理设置等）。</strong>在选择使用 <code>FastHttpUser</code> 时，需要权衡性能优势和功能支持。如果测试场景不需要大量并发用户，或者需要特定的 HTTP 功能，使用默认的 <code>HttpUser</code> 可能更合适。</p>
<p>以下是一个使用 <code>FastHttpUser</code> 的 Locust 测试脚本示例：</p>
<pre><code class="language-python">from locust import FastHttpUser, task, between

class MyFastHttpUser(FastHttpUser):
    wait_time = between(1, 5)

    @task
    def my_task(self):
      self.client.get("/")
</code></pre>
<h2 id="测试grpc等其他协议">测试<code>gRPC</code>等其他协议</h2>
<p><strong>locust 并非 http 接口测试工具 ， 只是内置了 “HttpUser” 示例 ，理论上来说，只要提供客户端，它可以测试任何协议。</strong></p>
<p>如果有测试 <code>gRPC</code>、<code>XML-RPC</code>、<code>requests-based libraries/SDKs</code>等需求，可以参考：<br>
https://docs.locust.io/en/stable/testing-other-systems.html</p>
<h2 id="其他">其他</h2>
<h4 id="主流性能测试工具对比">主流性能测试工具对比</h4>
<p>下面是 Locust、JMeter、Wrk 和 LoadRunner 四款性能测试工具的优缺点和支持的功能的对比表格：</p>
<table>
<thead>
<tr>
<th>工具名称</th>
<th>优点</th>
<th>缺点</th>
<th>支持的功能</th>
</tr>
</thead>
<tbody>
<tr>
<td>Locust</td>
<td>- 简单易用，支持 Python 语言<br>- 可以在代码中编写测试场景，灵活性高<br>- 可以使用分布式部署，支持大规模测试<br>- 支持 Web 和 WebSocket 测试</td>
<td>- 功能相对较少，不支持 GUI<br>- 对于非 Python 开发人员不太友好<br>- 在大规模测试时需要手动管理分布式节点</td>
<td>- HTTP(S)、WebSocket 测试<br>- 支持断言、参数化、数据驱动等功能<br>- 支持分布式测试</td>
</tr>
<tr>
<td>JMeter</td>
<td>- 功能丰富，支持多种协议<br>- 支持 GUI，易于使用<br>- 支持分布式部署，支持大规模测试<br>- 支持插件扩展，可以扩展功能</td>
<td>- 性能较差，不适合高并发测试<br>- 内存占用较高，需要较大的内存<br>- 学习曲线较陡峭</td>
<td>- HTTP(S)、FTP、JDBC、JMS、LDAP、SMTP、TCP、UDP 等多种协议的测试<br>- 支持断言、参数化、数据驱动等功能<br>- 支持分布式测试</td>
</tr>
<tr>
<td>Wrk</td>
<td>- 性能优异，支持高并发测试<br>- 支持 Lua 脚本编写，灵活性高<br>- 支持多种输出格式，方便结果分析</td>
<td>- 功能相对较少，不支持 GUI<br>- 只支持 HTTP 协议测试<br>- 学习曲线较陡峭</td>
<td>- HTTP(S) 测试<br>- 支持断言、参数化、数据驱动等功能</td>
</tr>
<tr>
<td>LoadRunner</td>
<td>- 功能丰富，支持多种协议<br>- 支持 GUI，易于使用<br>- 支持分布式部署，支持大规模测试<br>- 支持插件扩展，可以扩展功能</td>
<td>- 价格较高，不适合小型团队使用<br>- 学习曲线较陡峭<br>- 对于非 Windows 平台的支持不够友好</td>
<td>- HTTP(S)、FTP、JDBC、JMS、LDAP、SMTP、TCP、UDP 等多种协议的测试<br>- 支持断言、参数化、数据驱动等功能<br>- 支持分布式测试</td>
</tr>
</tbody>
</table>
<p>需要注意的是，这些工具的优缺点和支持的功能只是相对而言的，具体使用时需要根据实际需求和场景选择。</p>


</div>
<div id="MySignature" role="contentinfo">
    <div style="border: 1px solid #e4e4e4 !important; font-size: 13px; padding: 5px; padding-left: 20px">
<div style="text-align: left; font-weight: bold; color: #464545; font-family: -webkit-pictograph">文中可能存在描述不正确，欢迎大神们指正补充！</div>

<div style="text-align: left; font-weight: bold; color: #0eaf48; font-family: -webkit-pictograph">感谢阅读，如果觉得对你有帮助，就在右下角点个赞吧，感谢！</div>
</div>
<p style="padding: 10px; border-top: 1px solid #f1f0f0 !important; font-size: 15px; font-family: cursive">合抱之木，生于毫末；九层之台，起于累土；千里之行，始于足下。</p><br><br>
来源：https://www.cnblogs.com/Detector/p/17533341.html

查看完整版本: 【Python】万字长文，Locust 性能测试指北