使用locust测试MQTT协议

1.首先你本机需要安装python，因为locust是基于python环境，本次python的版本是3.9.2

pip3 install locust

# 查看版本
locust --version

from locust import User, TaskSet, events, task, between
import paho.mqtt.client as mqtt
import time
import random
COUNTClient = 0
#broker_address="broker.mqttdashboard.com"
broker_address="127.0.0.1"  #服务器ip地址，实际压测要根据mqtt的实际ip进行修改
REQUEST_TYPE = 'MQTT'
PUBLISH_TIMEOUT = 10000 #超时时间

def fire_locust_success(**kwargs):
    events.request_success.fire(**kwargs)

def increment():
    global COUNTClient
    COUNTClient = COUNTClient+1

def time_delta(t1, t2):
    return int((t2 - t1)*1000)

class Message(object):
    def __init__(self, type, qos, topic, payload, start_time, timeout, name):
        self.type = type,
        self.qos = qos,
        self.topic = topic
        self.payload = payload
        self.start_time = start_time
        self.timeout = timeout
        self.name = name

class PublishTask(TaskSet):
    def on_start(self):
        self.client.connect(host=broker_address, port=1883, keepalive=60)
        self.client.disconnect()

    @task(1)
    def task_pub(self):
        self.client.reconnect()
        self.client.loop_start()
        self.start_time = time.time()
        topic = "jmeter相关技术交流"  #主题名称，可根据需要进行修改
        payload = "Device - " + str(self.client._client_id)   #定义设备名称，可以按需更改
        count = random.randint(0,9)
        print('----执行随机数---{}'.format(count))  
       
        if count % 2 == 0:
  
           payload ='{"ctrl":{"mac":"60A423FFFE5E0589","endpoint":1,"OnOff":1, "ack":"You are Done"}}'
        else:
          payload ='{"ctrl":{"mac":"60A423FFFE5E0589","endpoint":1,"OnOff":0, "ack":"You are Done"}}'
        MQTTMessageInfo = self.client.publish(topic,payload,qos=0, retain=False)
        pub_mid = MQTTMessageInfo.mid
        print("Mid = " + str(pub_mid))   #用来在pycharm输出模拟的设备数
        self.client.pubmessage[pub_mid] = Message(
                    REQUEST_TYPE, 0, topic, payload, self.start_time, PUBLISH_TIMEOUT, str(self.client._client_id)
                    )
        MQTTMessageInfo.wait_for_publish()
        self.client.disconnect()
        self.client.loop_stop()
        time.sleep(5)

    wait_time = between(0.5, 10)

class MQTTLocust(User):
    tasks = {PublishTask}

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        increment()
        client_name = "Device - " + str(COUNTClient)
        self.client = mqtt.Client(client_name)
        self.client.on_connect = self.on_connect
        self.client.on_disconnect = self.on_disconnect
        self.client.on_publish = self.on_publish
        self.client.pubmessage  = {}

    def on_connect(client, userdata, flags, rc, props=None):
         fire_locust_success(
            request_type=REQUEST_TYPE,
            name='connect',
            response_time=0,
            response_length=0
            )

    def on_disconnect(client, userdata,rc,props=None):
        print("Disconnected result code "+str(rc))

    def on_publish(self, client, userdata, mid):
        end_time = time.time()
        message = client.pubmessage.pop(mid, None)
        total_time =  time_delta(message.start_time, end_time)
        fire_locust_success(
            request_type=REQUEST_TYPE,
            name=str(self.client._client_id),
            response_time=total_time,
            response_length=len(message.payload)
            )

1.在你的python脚本目录执行如下命令：

locust -f program_backups.py --host=127.0.0.1

2.命令解释 program_backups.py是你编写的python脚本文件名称，host是对应你的mqtt服务器ip，本次是在pycharm的命令行执行该命令

1.在浏览器输入http://localhost:8089/ ，进入locust压测页面 Number of users代表要压测的用户数，Spawn rate 代表每秒加载的用户数，Host是你的mqtt服务器ip 点击Start swarming开始压测

2.Statistics显示每个用户的请求数、响应时间等指标，Charts通过图表展示每秒请求数(RPS),响应时间，用户数 Failures显示报错数，Download Data压测报告

3.Locust界面不支持压测场景的设置，所以需要自己进行手动更改，比如压测完20用户，需要模拟100用户，就需要点击New test进入用户设置界面，修改完成，点击start swarming即可

4.最后贴一张实际工作过程中的压测结果，将并发数递度增至10、20、30时，当并发达20时，网关长时间无响应，再进一步增至30，网关完全无响应，得出网关的性能瓶颈在设备数为300、并发20达到拐点(本次是对网关进行压测)。