性能测试工具 深入浅出 Locust 实现

bugVanisher · 2020年06月17日 · 最后由 面对疾风吧 回复于 2020年10月26日 · 3189 次阅读

写在前面

本文将尝试通过一篇文章讲清楚开源压测工具——Locust 的原理和实现过程,聚焦在实现上,不拘泥在一堆源码中(本身 Locust 的源码就比较简单)。本人并不是 Locust 铁粉,只是对它的实现方式感兴趣,所以 jmeter、wrk 的粉丝们就不要来 battle 我了。在我看来,工具都有各自的优势和缺陷,区别在于使用的人,能否选择最合适的工具达到目的,当然你可以掌握所有的主流工具,对于特定的任务,哪个简单用哪个,也可以钟情于一个工具,信手拈来,只要你能在预期时间内完成也无可厚非。
网上有很多关于 Locust 源码实现的讲解,但是我觉得都太片面了,并且没有讲清楚,恰巧 Locust 在上个月(5 月)发布了 1.x 版本,整个重构力度还是蛮大的,因此我想通过这篇文章,介绍它最新的设计以及展示最全面的 Locust 实现,让大家感受到 Locust 的简单与优雅(整个 Locust 项目不过 4M 大小)。

一、架构与核心类

基本介绍

Locust 是开源、使用 Python 开发、基于事件、支持分布式并且提供 Web UI 进行测试执行和结果展示的性能测试工具。

Locust 的主要特性有两个:

  • 模拟用户操作:支持多协议,Locust 可以用于压测任意协议类型的系统
  • 并发机制:摒弃了进程和线程,采用协程(gevent)的机制,单台测试机可以产生数千并发压力

Locust 使用了以下几个核心库:

1) gevent
    gevent是一种基于协程的Python网络库,它用到Greenlet提供的,封装了libevent事件循环的高层同步API。
2) flask
    Python编写的轻量级Web应用框架。
3) requests
    Python Http库
4) msgpack-python
    MessagePack是一种快速、紧凑的二进制序列化格式,适用于类似JSON的数据格式。msgpack-python主要提供MessagePack数据序列化及反序列化的方法。
5) pyzmq
    pyzmq是zeromq(一种通信队列)的Python实现,主要用来实现Locust的分布式模式运行

系统架构及对比

我们知道,完整的压测系统应该包含以下组件:
压测系统

相比于主流的压测系统 LoadRunner 和 Jmeter,Locust 是一个更为纯粹的开源压测系统,框架本身的结构并不复杂,甚至只提供了最基础的组件,但也正因为如此,Locust 具有极高的可编程性和扩展性。对于测试开发同学来说,可以比较轻松地使用 Locust 实现对任意协议的压测。

工具简单对比

LoadRunner Jmeter Locust
压力生成器
负载控制器
系统资源监控器 x x
结果采集器
结果分析器

上表简单展示了几个工具包含的压测组件,Locust 的架构非常简单,部分组件的实现甚至都不完善,比如结果分析器,Locust 本身并没有很详细的测试报告。但这并不妨碍它成为优秀的开源框架。

Locust 的架构

Locust 最近发布了 1.x 版本,代码进行了重构,重封装,对很多类进行了重命名,本文尽量使用新名词,但涉及的旧名词不影响理解。

  • locust 架构上使用 master-slave 模型,支持单机和分布式
  • master 和 slave(即 worker)使用 ZeroMQ 协议通讯
  • 提供 web 页面管理 master,从而控制 slave,同时展示压测过程和汇总结果
  • 可选 no-web 模式(一般用于调试)
  • 基于 Python 本身已经支持跨平台

为了减少 CPython 的 GIL 限制,充分利用多核 CPU,建议单机启动 CPU 核数的 slave(多进程)。

核心类

上面展示的 Locust 架构,从代码层面来看究竟是如何实现的呢?下面我们就来窥视一番:

简单来说,Locust 的代码分为以下模块:

  • Runner-执行器,Locust 的核心类,定义了整个框架的执行逻辑,实现了 Master、Slave(worker)等执行器
  • User-压测用例,提供了 HttpUser 压测 http 协议,用户可以定义事务,断言等,也可以实现特定协议的 User
  • Socket-通信器,提供了分布式模式下 master 和 slave 的交互方式
  • RequestStats-采集、分析器,定义了结果分析和数据上报格式
  • EventHook-事件钩子,通过预先定义的事件使得我们可以在这些事件发生时 (比如 slave 上报) 做一些额外的操作
  • WebUI-提供 web 界面的操作台和压测过程展示

下面我们看看核心类的主要成员变量和方法

  • 用户定义的 User 类作为 Runner 的 user_classes 传入
  • master 的 client_listener 监听施压端 client 消息
  • slave 的 worker 方法监听 master 消息
  • slave 的 stats_reporter 方法上报压测数据,默认 3s 上报一次
  • slave 的 start 启动协程,使用 master 分配的并发数开始压测
  • slave 默认 1s 上报一次心跳,如果 master 超过 3s 未收到某个 slave 的心跳则会将其标记为 missing 状态
  • TaskSet 和 User 持有 client,可以在类中直接发起客户端请求,client 可以自己实现,Locust 只实现了 HttpUser

二、Runner 的状态与通信机制

接下来我们继续了解 Locust 分布式压测的核心:Runner 的状态和通信机制。我们知道 Locust 等压测工具支持分布式压测,就是说理论上可以通过不断添加压力机 (worker) 提高并发数量,这个机制让使用者可以自由地增减机器资源,从而达到期望的施压能力。

Runner 状态机

在分布式场景下,除了数据一致性,状态同步也是非常重要的。在 Locust 的 master-slave 架构下,需要管理 master 和 slave 的状态,不仅为了控制压测的开始或停止,也是为了掌握当前的压力机情况。那么都有哪些状态?

状态 说明
ready 准备就绪,master 和 slave 启动后默认状态
hatching 正在孵化压力机,对 master 来说正在告诉 slave 们开始干活,对 slave 来说是过渡状态,因为它们马上要 running
running 正在压测
cleanup 当发生 GreenletExit 时的状态,一般不会出现
stopping 表示正在通知 slave 们停止,只有 master 有这个状态
stopped 压测已经停止
missing 状态丢失,只有 slave 有的状态,默认 3 秒如果 master 没有收到 slave 的心跳就会认为它 missing 了,一般是进程没有正常退出导致

Runner 的状态不多,但是在压测过程中起到非常重要的作用,状态之间是按约定的方式进行扭转的,我们使用 Locust 的 web 界面管理 master 的状态,master 根据我们的操作通过通信机制推进 slave 的状态。

通信机制

Master 与 Slave 之间是通过 Zeromq 建立的 TCP 连接进行通信的(一对多)。

ZeroMQ(简称 ZMQ)是一个基于消息队列的多线程网络库,其对套接字类型、连接处理、帧、甚至路由的底层细节进行抽象,提供跨越多种传输协议的套接字。

ZMQ 是网络通信中新的一层,介于应用层和传输层之间(按照 TCP/IP 划分),其是一个可伸缩层,可并行运行,分散在分布式系统间。

master 与各个 slave 各维持一个 TCP 连接,在每个连接中,master 下发的命令,slave 上报的信息等自由地的传输着。

消息格式

class Message(object):
    def __init__(self, message_type, data, node_id):
        self.type = message_type
        self.data = data
        self.node_id = node_id

    def serialize(self):
        return msgpack.dumps((self.type, self.data, self.node_id))

    @classmethod
    def unserialize(cls, data):
        msg = cls(*msgpack.loads(data, raw=False))
        return msg

其中 message_type 指明消息类型,data 是实际的消息内容,node_id 指明机器 ID。Locust 使用msgpack做序列化与反序列化处理。

消息类型和结构

master 和 slave 之间的消息类型不过 10 种,其中大部分消息由 slave 上报给 master,下方表格可以清楚的看到,实现一套分布式系统并没有那么复杂。

序号 message_type 发送者 data 格式 发送时机
1 client_ready slave slave 启动后或压测停止完成
2 hatching slave 接受到 master 的 hatch 先发送一个确认
3 hatch_complete slave {"count":n} 所有并发用户已经孵化完成
4 client_stopped slave 停止所有并发用户后
5 heatbeat slave {"state": x,"current_cpu_usage":x} 默认每 3 秒上报一次心跳
6 stats slave {"stats":[], "stats_total":{}, "errors":{},"user_count":x} 每 3 秒上报一次压测信息
7 exception slave {"msg":x, "traceback":x} TaskSet.run 出现异常
8 hatch master {"hatch_rate":x, "num_users":x, "host":x, "stop_timeout":x} 开始 swarm
9 stop master 点击 stop
10 quit master,slave 手动或其他方式退出的时候

可以看到上面有一种非常重要的消息类型——stats,压测的结果采集都封装在这个消息里。

三、结果采集器

统计对象

从上面我们知道有一个非常重要的消息类型——stats,这个是 slave 给 master 发送的消息,默认每 3 秒钟上报一次。stats 消息的结构如下所示:

{
    "stats": [],
    "stats_total": {},
    "errors": {},
    "user_count": 10
}

实际上,slave 也是持有着类似如上 json 格式的三个对象:

  • RequestStats
  • StatsEntry
  • StatsError

其中每一个 locust 进程会维护一个全局 RequestStats 单例 global_stats,这个实例包含一个 StatsEntry 实例 total(对应 json 的 stats_total),以及两个字典 entries(对应 json 的 stats)和 errors(对应 json 的 errors),其中 entries 字典 key 为 (name,method),对应值为一个 StatsEntry 实例,另一个 errors 字典的 key 为 (name,method,error),对应值为一个 StatsError 实例。可见,global_stats 包含了单个 slave 汇总的信息,以及各个请求 url 或 name 的统计信息。在分布式的场景下,每个 slave 都会维护一个 global_stats, 在一个上报周期达到后将信息发送到 master,发送完成后就会重置所有数据,进入下一个周期的统计。

上图中红色的字段是 slave 真正上报给 master 的数据。

统计过程

那么 slave 是如何统计消息,又究竟需要上报什么内容给 master?master 又是如何汇总的呢?下面我们来看看整个统计过程:

在每一次请求后,无论是成功还是失败,都会触发对应的 request_success 或者 request_failure 事件,stats.py 文件中的数据统计模块订阅了对应的事件,会调用 global_stats 对数据进行统计。

在 slave 的一个上报周期达到后,触发 on_report_to_master 事件,此时 global_stats 就会依次调用以下方法对数据进行序列化:

  • serialize_stats
  • total.get_stripped_report
  • serialize_errors

其实也就是对上面提到的 total 和两个字典中的内容进行序列化,其实就是转为 json 字符串。

def on_report_to_master(client_id, data):
    data["stats"] = global_stats.serialize_stats()
    data["stats_total"] = global_stats.total.get_stripped_report()
    data["errors"] =  global_stats.serialize_errors()
    global_stats.errors = {}

下图是断点过程看到的 stats 消息内容(在 msgpack 序列化之前):

每秒请求数和响应时间及其对应请求个数

StatsEntry 有两个比较重要的对象,分别是 num_reqs_per_sec 和 response_times,它们都是字典类型,其中 num_reqs_per_sec 的 key 是秒时间戳,显示当前秒完成了多少个请求,统计的时间是完成请求的时刻,比如如果一个请求从第 0 秒开始,第 3 秒完成,那么这个请求统计在第 3 秒的时间戳上,这个对象可以很方便的计算出 rps。response_times 的 key 是响应时间,单位是豪秒,为了防止 key 过多,做了取整,比如 147 取 150, 3432 取 3400 和 58760 取 59000,这个是为了方便获得类似 90% 请求的完成时间 (小于等于该时间),99% 请求的完成时间,下面具体的源码:

def _log_time_of_request(self, current_time):
    t = int(current_time)
    self.num_reqs_per_sec[t] = self.num_reqs_per_sec.setdefault(t, 0) + 1
    self.last_request_timestamp = current_time

def _log_response_time(self, response_time):
    if response_time is None:
        self.num_none_requests += 1
        return

    self.total_response_time += response_time

    if self.min_response_time is None:
        self.min_response_time = response_time

    self.min_response_time = min(self.min_response_time, response_time)
    self.max_response_time = max(self.max_response_time, response_time)

    # to avoid to much data that has to be transfered to the master node when
    # running in distributed mode, we save the response time rounded in a dict
    # so that 147 becomes 150, 3432 becomes 3400 and 58760 becomes 59000
    if response_time < 100:
        rounded_response_time = response_time
    elif response_time < 1000:
        rounded_response_time = int(round(response_time, -1))
    elif response_time < 10000:
        rounded_response_time = int(round(response_time, -2))
    else:
        rounded_response_time = int(round(response_time, -3))

    # increase request count for the rounded key in response time dict
    self.response_times.setdefault(rounded_response_time, 0)
    self.response_times[rounded_response_time] += 1

master 汇总信息

slave 的每一个 stats 消息到达 master 后,都会触发 master 的 slave_report 事件,master 也拥有自己的 global_stats,因此只需要将对应的信息进行累加(可以理解是所有 slave 对应内容的汇总)。具体在 StatsEntry 的 extend 方法:

def extend(self, other):
        """
        Extend the data from the current StatsEntry with the stats from another
        StatsEntry instance. 
        """
        if self.last_request_timestamp is not None and other.last_request_timestamp is not None:
            self.last_request_timestamp = max(self.last_request_timestamp, other.last_request_timestamp)
        elif other.last_request_timestamp is not None:
            self.last_request_timestamp = other.last_request_timestamp
        self.start_time = min(self.start_time, other.start_time)

        self.num_requests = self.num_requests + other.num_requests
        self.num_none_requests = self.num_none_requests + other.num_none_requests
        self.num_failures = self.num_failures + other.num_failures
        self.total_response_time = self.total_response_time + other.total_response_time
        self.max_response_time = max(self.max_response_time, other.max_response_time)
        if self.min_response_time is not None and other.min_response_time is not None:
            self.min_response_time = min(self.min_response_time, other.min_response_time)
        elif other.min_response_time is not None:
            # this means self.min_response_time is None, so we can safely replace it
            self.min_response_time = other.min_response_time
        self.total_content_length = self.total_content_length + other.total_content_length

        for key in other.response_times:
            self.response_times[key] = self.response_times.get(key, 0) + other.response_times[key]
        for key in other.num_reqs_per_sec:
            self.num_reqs_per_sec[key] = self.num_reqs_per_sec.get(key, 0) + other.num_reqs_per_sec[key]
        for key in other.num_fail_per_sec:
            self.num_fail_per_sec[key] = self.num_fail_per_sec.get(key, 0) + other.num_fail_per_sec[key]

核心指标

Locust 核心的指标其实就 4 个:

  • 并发数
  • RPS
  • 响应时间
  • 异常率

我们还是回到文章开头的那个 json:

{
    "stats": [],
    "stats_total": {},
    "errors": {},
    "user_count": 10
}

结合上方 Locust 的压测过程截图,我们可以看到,各个接口的指标其实就是 stats 对象里的各个字段,而最下方汇总的 Aggregated 这一行则对应 stats_total 的各个字段,尽管这个 json 只是 slave 单个 stats 消息的内容,却也是最终要显示的内容,只是 master 对各个消息做了汇总而已。汇总的方式也相当简单,请见上方的 StatsEntry 的 extend 方法。

因为 master 和 web 模块是一起部署的,因此 web 可以直接使用 master 的 global_stats 对象并展示其内容,可以做到动态显示。

本文讲解的 eventHook 基于 Locust 的 1.x 版本。

四、事件钩子

Locust 中的事件钩子,我认为设计得非常巧妙,所以最后再来讲一下它。event.py 模块包含了两个类,一个是事件钩子定义类 EventHook,一个是事件钩子类型类 Events,为不同的事件提供 hook。事件处理函数注册相应的 hook 以后,我们可以很方便的的基于 event 触发处理函数,实现事件驱动。

EventHook 定义了三个方法:

def add_listener(self, handler):
    self._handlers.append(handler)
    return handler

def remove_listener(self, handler):
    self._handlers.remove(handler)

def fire(self, *, reverse=False, **kwargs):
    if reverse:
        handlers = reversed(self._handlers)
    else:
        handlers = self._handlers
    for handler in handlers:
        handler(**kwargs)

可以看到 add_listener、remove_listener 的作用是注册或删除监听函数,fire 方法是触发处理函数。EventHook 的实现相比 Locust 0.x 版本有较大改变(不再使用 Python 的内置魔方方法)。

Events 中包含了 11 个事件钩子,分别是:

image-20200616152110342

事件钩子的实现原理

事件钩子的原理可以简单理解成,1.定义处理函数 ——> 2.注册到某个 eventHook ——> 3.在某个时机触发 eventHook ——> 4.该 eventHook 遍历执行所有处理函数。在代码层面就是定义函数,然后 add_listener,最后在想要的位置触发 eventHook 的 fire 方法并传入定义好的参数,这里参数是固定的,不能随意传入,之所以每个处理函数都能对参数进行修改,是因为这里的参数传递是『引用传递』,明白了这一点就算是掌握了 EventHook 的基本原理。

其中 Locust 本身预定义并注册了一些处理函数,比如事件钩子 report_to_master、worker_report 都有对应的处理函数,实现分布式模式下数据上报时数据的构造和计算,比如事件钩子 init,初始化 master 的 WebUI。

事件钩子的作用

那么,事件钩子究竟有什么作用?

在我看来有以下作用:

  • 代码解耦,比如解耦 Runner 和 Stats 模块。
  • 提供扩展性,通过预置的钩子和触发点,为使用者提供了可扩展的特性。

举扩展性的例子,使用者可以很轻松地:

  • 往 worker 上报的 stats 消息添加自定义数据
  • 在 Locust 实例启动时在 WebUI 中添加自定义接口
  • 在每次请求成功或失败之后做一些额外的事情

如何使用钩子

在 Locust 1.x 版本之前,使用以下方法定义和注册钩子:

def on_report_to_master(client_id, data):
    data["stats"] = global_stats.serialize_stats()
    data["stats_total"] = global_stats.total.get_stripped_report()
    data["errors"] =  global_stats.serialize_errors()
    global_stats.errors = {}

def on_slave_report(client_id, data):
    for stats_data in data["stats"]:
        entry = StatsEntry.unserialize(stats_data)
        request_key = (entry.name, entry.method)
        if not request_key in global_stats.entries:
            global_stats.entries[request_key] = StatsEntry(global_stats, entry.name, entry.method)
        global_stats.entries[request_key].extend(entry)
        ...

events.report_to_master += on_report_to_master
events.slave_report += on_slave_report

在 1.x 及之后的版本有两种使用方式:

# 方式一与之前类似
def on_report_to_master(client_id, data):
    data["stats"] = stats.serialize_stats()
    data["stats_total"] = stats.total.get_stripped_report()
    data["errors"] =  stats.serialize_errors()
    stats.errors = {}

def on_worker_report(client_id, data):
    for stats_data in data["stats"]:
        entry = StatsEntry.unserialize(stats_data)
        request_key = (entry.name, entry.method)
        if not request_key in stats.entries:
            stats.entries[request_key] = StatsEntry(stats, entry.name, entry.method, use_response_times_cache=True)
        stats.entries[request_key].extend(entry)

    for error_key, error in data["errors"].items():
        if error_key not in stats.errors:
            stats.errors[error_key] = StatsError.from_dict(error)
        else:
            stats.errors[error_key].occurrences += error["occurrences"]

    stats.total.extend(StatsEntry.unserialize(data["stats_total"]))

events.report_to_master.add_listener(on_report_to_master)
events.worker_report.add_listener(on_worker_report)

# 方式二,使用装饰器
@user2ter.add_listener
def on_report_to_master(client_id, data):
    """
    This event is triggered on the worker instances every time a stats report is
    to be sent to the locust master. It will allow us to add our extra content-length
    data to the dict that is being sent, and then we clear the local stats in the worker.
    """
    data["content-length"] = stats["content-length"]
    stats["content-length"] = 0

@events.worker_report.add_listener
def on_worker_report(client_id, data):
    """
    This event is triggered on the master instance when a new stats report arrives
    from a worker. Here we just add the content-length to the master's aggregated
    stats dict.
    """
    stats["content-length"] += data["content-length"]

第二种方式更加简洁,不容易忘记注册处理函数。

写在最后

首先需要声明的是,本人并不是性能测试方向的专家,只是做过一些性能测试,捣鼓过一些工具,之所以写出这篇文章,初衷只是为了在组内分享。了解 Locust 的人,很少会真的用他来进行大型业务场景压测,一是它的压测性能受限,二是它没有很好的测试报告,现在这些在我看来已经不是问题,我想在后面分两篇文章介绍,敬请期待:

一、如何提高 Locust 的压测性能——boomer
二、重新定义 Locust 的测试报告

共收到 21 条回复 时间 点赞

看到您有关 Locust 的内容好激动,内容和标题很是契合,有大量干货内容值得咀嚼消化。因为之前使用过 0.14 之前的版本,也有利用它在公司产品的线上线下做过一些相关测试,感觉很不错,很 pythonic。之后接触到 httprunner 也用到了它,实现接口功能和性能测试一套 case 解决。最后非常感兴趣和会继续关注您后续两篇文章的分享。。。

讲得很好,期待后续的两篇.

老哥,这篇文章写得太丰富了

一看就是走心了👏

期待 goroutine

好文 ,,,坐等后续两篇

单数 回复

哈哈,感觉是挖了个坑給自己

ola嘿嘿 回复

社区的第一篇文章,除了有点长,还是有诚意的。。。

stevenxu 回复

我的博客分了几篇来写,这里我做了整合,没有必要一篇篇来了,一次看个过瘾吧,,

an footman 回复

纯 Python 压测会給自己带来很多麻烦,不建议在实际业务中这么做哈,接下来的两篇文章感觉是给我自己挖坑了😂 ,我尽量写出点干货吧,哈哈

locust 参数化的时候读取文件的内容来加到队列里,然后运行 task 时从队列中取。分布式的情况下,需要每个参数化文件中的数据都不一样吗

Yh。 回复

嗯嗯,分布式模式下每个 slave(worker)都是独立的,如果你的文件是一样的那就会重复。文件中的数据是否一样,取决于你的需求场景啊,如果必须保证全局唯一,方法是有很多的,我提供一个比较『骚』的思路:
整体思想就是每个 slave 在压测前去获取一批全局唯一的数据
1、利用 init 事件钩子在 master 注册一个新的 API,负责分发唯一数据。
2、同样利用钩子在 slave 压测前去 master 获取并初始化全局唯一数据,然后压测。

写的很详细,我也正好学习学习

刚好这几天在看 locust😀 期待后续文章

感觉很理论,不知道我想错没有

剩下的两篇啥时候出啊

bugVanisher 如何提高 Locust 的压测性能 中提及了此贴 07月21日 21:30

locust 网页上查看统计数据和图表曲线的体验还是很好的,美中不足的是他的导出功能太弱了。
我为 locust 写了一个 “导出报告” 的功能,可一键下载整份图文报告。
https://github.com/locustio/locust/pull/1516
目前该功能已被官方收录,下一个版本的 locust 将可直接导出报告

bugVanisher 回复

为啥不建议在实际业务中使用,个人不太喜欢 LR 和 jmeter,想用 locust 在实际业务中替代,最近也是在学习这个。

测试zzZ 回复

麻烦看完我的系列文章,相信你会有自己的答案😀

迄今为止看到介绍最详细的了,赞!

需要 登录 后方可回复, 如果你还没有账号请点击这里 注册