压力测试(QPS)及测试工具Locust-Toy模板网

这篇具有很好参考价值的文章主要介绍了压力测试(QPS)及测试工具Locust。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

压力测试：

通常指的是确定接口或服务能够处理的最大请求量（吞吐量）和并发用户数，同时保持合理的响应时间和稳定性。

性能目标

最大吞吐量：系统每秒可以处理的请求数。
最大并发用户数：系统可以同时支持的用户数。
响应时间：系统响应请求的时间。
稳定性和可靠性：在高负载下系统的稳定性。

QPS( Queries Per Second，即每秒查询数)

在进行性能测试时，通常会逐步增加并发用户数（或请求频率），以便找到接口或系统的最大QPS值。这个过程中，您会观察到系统的响应时间和请求失败率随着并发量的增加而变化。确定QPS的最大值通常涉及到平衡性能指标和系统的稳定性。

如何确定QPS的最大值

响应时间：随着请求量的增加，响应时间可能会增长。您需要确定一个可接受的最大响应时间阈值。当响应时间超过这个阈值时，即使系统还能处理更多请求，也不应该继续增加负载。
错误率：随着负载的增加，错误率（包括超时、5XX状态码等）可能会上升。通常，您需要设定一个错误率阈值（例如1%），当错误率超过这个阈值时，可以认为已经达到了系统的性能极限。
资源利用率：监控系统的资源利用率（如CPU、内存、网络带宽等）也很重要。资源饱和可能导致性能下降和错误率上升。资源利用率接近或达到100%通常意味着接近系统极限。

综上考虑确定最大QPS：当响应时间超过预定阈值、错误率升高或资源利用率达到饱和时，即认为找到了系统的最大QPS。这是系统在保持良好性能和稳定性的前提下能够处理的最高请求数。

计算QPS

QPS 的计算公式相对简单：

QPS = 总请求次数 / 请求持续时间(秒)

例如，如果一个服务在 10 秒内处理了 2000 个请求，那么：

QPS = 2000 / 10 = 200

这意味着该服务每秒平均处理 200 个请求。

在压力测试中，1个用户每秒请求10000次，与10000个用户每秒请求1次，有什么不同吗？

虽然在表面上看似都产生了每秒 10000 次的请求量，但实际对系统的影响和测试的含义有显著不同：

1. 并发性和资源使用

1个用户请求 10000 次：这种情况下，请求是顺序发出的，即使以非常快的速度，也只有一个请求的处理在进行。这种情况测试的是服务对单一用户的高频率请求的处理能力。服务器资源（如数据库连接）可能主要集中在处理这一个用户的请求上，不一定能反映出在多用户并发情况下的资源竞争和分配问题。
10000个用户各请求1次：这种情况下，请求是并发发出的，服务器需要同时处理来自一万个不同用户的请求。这种情况测试的是服务的并发处理能力和系统资源（如数据库连接池、内存、CPU）在大量并发请求下的表现。这更接近真实世界的使用场景，能够暴露出服务在高并发下可能遇到的性能瓶颈和问题

2. 会话和状态管理

在一个用户高频请求的场景中，如果服务涉及到会话或状态管理，那么这些状态可能会在连续的请求之间保持，影响测试结果。
在多用户并发请求的场景中，服务需要管理和维护来自多个用户的会话和状态，这可能揭示出在会话管理、状态一致性等方面的问题。

3. 网络和负载平衡

单用户高频请求可能无法充分测试网络的负载均衡和连接限制，因为所有请求可能都通过相同的网络路径和连接进行。
多用户并发请求则可能涉及到更多的网络路径和连接，能够测试负载均衡器在分配请求到不同服务器或服务实例时的效果，以及服务在处理来自不同源的请求时的网络性能。

虽然两种情况下每秒的请求量相同，但从测试的目的和重点来看，它们反映了系统不同方面的性能和扩展性。多用户并发请求的测试更能反映真实世界中服务的并发处理能力和资源管理效率，因此在进行压力测试和性能评估时，应该考虑到测试场景与真实使用场景的匹配度。

压力测试工具Locust

一个用 Python 编写的开源负载测试工具，易于编写测试脚本，支持模拟数百万用户并发。

1. 安装 Locust

pip install locust

2. 编写 Locust 测试脚本

创建一个名为 locustfile.py 的 Python 文件，并在文件中定义用户行为和任务。以下是一个基本的测试脚本示例，它模拟用户访问一个 HTTP API：

from locust import HttpUser, between, task
import json
import base64

global pic_base64
img_path = './ocr/test_images/1.jpg'
with open(img_path,'rb') as pic:
    pic_base64 = str(base64.b64encode(pic.read()),encoding='utf-8')


class WebsiteUser(HttpUser):
    host = "http://localhost:8080"
    wait_time = between(1, 5)

    @task
    def ocr_task(self):
        # data = {"img_base64": pic_base64}
        data = {"key": ["image"], "value": [pic_base64]}
        headers = {'Content-Type': 'application/json'}
        self.client.post("/ocr/prediction", data=json.dumps(data), headers=headers)