基于Locust实现MQTT协议服务的压测脚本-六虎

最近在忙事务的间隙，穿插着做了些功能测验。

一、背景简介

事务背景大约介绍一下，便是依照国标规定，车辆需求上传一些指定的数据到ZF的指定渠道，一起车辆也会把数据传到企业云端服务上，所以乎就发生了一些功能需求。

现在咱们仅仅先简略的进行了一个功能场景的测验，便是评价现在服务是否能够支撑，预期的最大一起在线车辆上传数据。通过评价，在线车辆数据依照预期的10倍来进行的，而且后边添加持续运转12h检查服务链路的安稳性。

本篇并不是一个严谨的功能测验过程成果共享，主要是共享下关于mqtt协议服务的压测脚本的编写。因为之前我也没触摸过MQTT协议的压测，网上关于相关的压测脚本的内容也比较杂乱，所以记载一下，仅供参考。

捋一下链路就知道需求生成哪些数据（因为服务还未上线运用，所以发生的压测数据后边能够直接整理掉即可。）：

一些前置数据：比方数据库、缓存里涉及到的车辆数据，通讯秘钥数据等等，这些能够之前写脚本一次性生成即可。
车辆上报的数据：车辆上报到云端的数据，是通过一系列加密转码，期间还要设计到解密等，这个通过评价，能够简化其间的某些环境，所以所有的车能够直接发送相同的数据即可。
车辆数据：最终便是生成对应的车辆数据，一起在线，依照评价的频率发送数据。

其间第1、2的数据在之前针对性的别离生成即可，第3步的车辆发送数据便是压测脚本要干的工作了。

二、技术选型

这个却是很快，搜索引擎大约搜了一下，内容很少，或者说对我有用的内容很少。有看到jmeter有相关插件的，但是这个方案基本上我都是否决的，一来我不擅长用，而来我觉得用起来肯定会比自己编码要费事的多。

所以就持续编码好了，依然首选python，想到了locust库，后来看官方文档的时分，看到locust也针对mqtt协议拓宽了一些内容。但是我测验下来不太符合我这的需求，也可能当时我用的不对吧，所以就只能自己来从零开端编写了。

搜索中又发现Python中用于mqtt协议的库叫paho.mqtt，支撑衔接署理，音讯的订阅、收发等等，所以最终确认运用：locust+paho.mqtt的组合来完成本次的负载脚本。

三、代码编写

1. 脚本代码

暂时没做代码分层，现在场景简略，就直接都放一个模块里了，有点长，先贴上来，后边部分会对脚本的要点内容进行拆解。

脚本现在做了这些工作：

从db中查询有用可用的所有测验车辆信息数据
依据命令行的输入参数，指定发动的车辆数，以及与broker署理树立衔接的频率
树立衔接成功的车辆，就能够依据脚本里指定的频次，来像broker发送数据
脚本核算衔接数、恳求数、呼应时刻等信息写到报表中
调试遇到车辆会批量断开衔接的状况，添加了当车辆断开衔接时，把断开时刻、车辆信息写到本地csv中，便利第二天来检查分析。

import csv
import datetime
import queue
import os
import sys
import time
import ssl
from paho.mqtt import client as mqtt_client
# 依据不同系统进行路径适配
if os.name == "nt":
    path = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
    sys.path.insert(0, path)
    from GB_test.utils.mysql_operating import DB
elif os.name == "posix":
    sys.path.append("/app/qa_test_app/")
    from GB_test.utils.mysql_operating import DB
from locust import User, TaskSet, events, task, between, run_single_user
BROKER_ADDRESS = "broker服务地址"
PORT = 1111
PASSWORD = "111111"
PUBLISH_TIMEOUT = 10000  # 超时时刻
TEST_TOPIC = "test_topic"
TEST_VALUE = [16, 3, -26, 4, 0, 36,.......]  # 用来publish的测验数据，仅示意
BYTES_DATA = bytes(i % 256 for i in TEST_VALUE)  # 事务需求转换成 byte 类型后再发送
# 创建行列
client_queue = queue.Queue()
# 衔接DB，读取车辆数据
db = DB("db_vmd")
select_sql = "select xxxx"  
client_list = db.fetch_all(select_sql)
print("车辆数据查询完毕，数据量：{}".format(len(client_list)))
for t in client_list:
    # 把可用的车辆信息存到行列中去
    client_queue.put(t)
def fire_success(**kwargs):
    """恳求成功时调用"""
    events.request.fire(**kwargs)
def calculate_resp_time(t1, t2):
    """核算呼应时刻"""
    return int((t2 - t1) * 1000)
class MQTTMessage:
    """已发送的音讯实体类"""
    def __init__(self, _type, qos, topic, payload, start_time, timeout):
        self.type = _type,
        self.qos = qos,
        self.topic = topic
        self.payload = payload
        self.start_time = start_time
        self.timeout = timeout
# 核算总共发送成功的音讯数量
total_published = 0
disconnect_record_list = []  # 界说寄存衔接断开的记载的列表容器
class PublishTask(TaskSet):
    @task
    def task_publish(self):
        self.client.loop_start()
        topic = TEST_TOPIC
        payload = BYTES_DATA
        # 记载发送的开端时刻
        start_time = time.time()
        mqtt_msg_info = self.client.publish(topic, payload, qos=1, retain=False)
        published_mid = mqtt_msg_info.mid
        # 将发送成功的音讯内容，放入client实例的 published_message 字段
        self.client.published_message[published_mid] = MQTTMessage(REQUEST_TYPE,
                                                                   0,
                                                                   topic,
                                                                   payload,
                                                                   start_time,
                                                                   PUBLISH_TIMEOUT)
        # 发送成功回调
        self.client.on_publish = self.on_publish
        # 断开衔接回调
        self.client.on_disconnect = self.on_disconnect
    @staticmethod
    def on_disconnect(client, userdata, rc):
        """ broker衔接断开，放入列表容器"""
        disconnected_info = [str(client._client_id), rc, datetime.datetime.now()]
        disconnect_record_list.append(disconnected_info)
        print("rc状况:{} - -".format(rc), "{}-broker衔接已断开".format(str(client._client_id)))
    @staticmethod
    def on_publish(client, userdata, mid):
        if mid:
            # 记载音讯发送成功的时刻
            end_time = time.time()
            # 从已发送的音讯容器中，取出音讯
            message = client.published_message.pop(mid, None)
            # 核算开端发送到发送成功的耗时
            publish_resp_time = calculate_resp_time(message.start_time, end_time)
            fire_success(
                request_type="p_success",
                name="client_id: " + str(client._client_id),
                response_time=publish_resp_time,
                response_length=len(message.payload),
                exception=None,
                context=None
            )
            global total_published
            # 成功发送累加1
            total_published += 1
class MQTTLocustUser(User):
    tasks = [PublishTask]
    wait_time = between(2, 2)
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 从行列中获取客户端 username 和 client_id
        current_client = client_queue.get()
        self.client = mqtt_client.Client(current_client[1])
        self.client.username_pw_set(current_client[0], PASSWORD)
        # self.client.username_pw_set(current_client[0] + "1", PASSWORD)  # 模仿client衔接报错
        # 界说一个容器，寄存已发送的音讯
        self.client.published_message = {}
    def on_start(self):
        # 设置tls
        context = ssl.SSLContext(ssl.PROTOCOL_TLS)
        self.client.tls_set_context(context)
        self.client.connect(host=BROKER_ADDRESS, port=PORT, keepalive=60)
        self.client.on_connect = self.on_connect
    def on_stop(self):
        print("publish 成功, 当时已成功发送数量：{}".format(total_published))
        if len(disconnect_record_list) == 0:
            print("无断开衔接的client")
        else:
            # 把断开记载里的信息写入csv
            with open("disconnect_record.csv", "w", newline='', encoding='UTF8') as csvfile:
                writer = csv.writer(csvfile)
                writer.writerow(['client_id', 'rc_status', 'disconnected_time'])
                for i in disconnect_record_list:
                    writer.writerow(i)
            print("断开衔接的client信息已写入csv文件")
    @staticmethod
    def on_connect(client, userdata, flags, rc, props=None):
        if rc == 0:
            print("rc状况:{} - -".format(rc), "{}-衔接broker成功".format(str(client._client_id)))
            fire_success(
                request_type="c_success",
                name='count_connected',
                response_time=0,
                response_length=0,
                exception=None,
                context=None
            )
        else:
            print("rc状况:{} - -".format(rc), "{}-衔接broker失利".format(str(client._client_id)))
            fire_success(
                request_type="c_fail",
                name="client_id: " + str(client._client_id),
                response_time=0,
                response_length=0,
                exception=None,
                context=None
            )
if __name__ == '__main__':
    run_single_user(MQTTLocustUser)

2. 代码分析-locust库部分

并发恳求才能仍是运用的locust库的才能。官方只提供了http协议接口的相关类，没直接提供mqtt协议的，但是咱们能够依照官方的规范，自界说相关的类，只要继承User和TaskSet即可。

User类

首先是先界说User类，这儿便是用来生成我要用来测验的车辆。

类初始化的时分，黄色框里，会去行列里取出车辆信息，用来做一些相关的设置。client来源于from paho.mqtt import client as mqtt_client提供的才能，固定用法，依照人家的文档运用就行。

红色框里，是User类的2个重要了解特点：

tasks: 这儿界说了生成的用户需求去干哪些工作，也便是对应脚本里的PublishTask类下面界说的内容。
wait_time: 用户在履行task时刻隔逗留的时刻，能够是个区间，在里边随机。我这儿意思是每2s发送一次数据到broker。

绿色框里，界说了一个字典容器，用来寄存当时用户已发送成功的音讯内容，因为后边我要取出来把里边相关的数据写到生成的报表中去。

蓝色框里有2个办法，也是locust提供的才能：

on_start：当用户开端运转时调用，这儿我做了车辆衔接broker署理的处理，留意这儿需求设置tls，因为服务衔接需求。
on_stop：当用户完毕运转时调用，这儿我做了一些其他的处理，比方把运转期间断开衔接的车辆信息写到本地csv中。

TaskSet类

界说好User类，就需求来界说TaskSet类，你得告诉发生出来的用户，要干点啥。

我这依据事务需求，便是让车辆不停的像broker发送数据即可。

红色部分，同样是paho.mqtt提供的才能，会发动新的线程去履行你界说的工作。

黄色部分，便是做发送数据的操作，而且我能够拿到一些回来，检查源码就能够知道回来的是MQTTMessageInfo类。

留意回来的2个特点：

mid: 回来这个音讯发送的顺序
rc: 表明发送的呼应状况，0 便是成功

绿色部分，还记得我在上面的User类中界说了一个容器，在这儿就把发送的音讯相关信息放到容器中去，留着后边运用。

2. 代码分析-paho.mqtt库部分

上面的代码现已用到了不少paho.mqtt的才能，这儿再进行整体梳理下。

client.Client():声明一个client
client.username_pw_set(): 设置客户端的用户名，暗码
client.tls_set_context：设置ssl模式
client.connect(): 衔接署理
client.publish:向署理推送音讯

还用到了一些回调函数：

on_connect：衔接操作成功时回调
on_publish：发布成功时回调
on_disconnect：客户端与署理断开衔接时回调

别的还用到了一个事件函数events.request。

当客户端发送恳求时会调用，不管是恳求成功仍是恳求失利；当我需求自界说我的报告内容时，就需求用到这个event。

检查源码，知道里边要传哪些参数，那咱们在调用时分就需求传入对应的参数。

比方我在发送回调函数里调用了该办法。

所以最终在控制台显示的报告里就有我界说的内容了。

因为后来在运用中发现，不知道会在什么时分呈现批量断开的状况，所以在on_disconnect回调函数里添加了对应处理，把相关的断开信息记载下来，运转完毕的时分写到本地文件里去。

后来我主动测验客户端断开的状况测验了下文件的写入成果，功能正常。

三、小结

后边就开端运转了，在运转过程中，开发重视链路服务的各项指标，这儿就不展开了，事务缠身就并没有过多的去做这个工作，何况也不专业。的确也发现了不少问题，后边逐步优化，再持续测验。

现在安稳运转12h，服务正常，暂时就先告一段落了。后边还有会相关其他功能测验场景，到时就能够针对性的展开共享下了。

别的，这个脚本共享也仅仅仅供参考，现在我这是运用简略，本着能用就行，可能存在一些不合理需求优化的当地，有需求的朋友还请自行查阅相关文档。

基于Locust实现MQTT协议服务的压测脚本

一、背景简介

二、技术选型

三、代码编写

1. 脚本代码

2. 代码分析-locust库部分

2. 代码分析-paho.mqtt库部分

三、小结

作者信息

推广

基于Locust实现MQTT协议服务的压测脚本

一、背景简介

二、技术选型

三、代码编写

1. 脚本代码

2. 代码分析-locust库部分

2. 代码分析-paho.mqtt库部分

三、小结

相关文章

Keras深度学习——使用预训练VGG16模型实现性别分类

开源框架 NanUI 作者转行卖钢材,谈生存~

Jetpack Compose Navigation 中的参数传递

字节开源安卓开发利器-CodeLocator

作者信息

推广