愿君学长松,慎勿作桃李。这篇文章主要讲述Locust如何测试物联网MQTT相关的知识,希望能为你提供帮助。
MQTT是干什么的简单来说,它是物联网的通信协议,是消息通道建立,消息发送和消息订阅的标准。如果大家想了解更多概念上的详细可以网上搜索。
Locust测试MQTT的步骤测试步骤,可以用以下图形表示:
准备环境,安装Locust测试环境
这个比较简单,主要是准备好python的虚拟开发环境,并安装好locust的python软件包。可以在Locust官方网站找到相关步骤,这里不再赘述。
安装MQTT客户端库 paho-mqtt
可以在Python虚拟环境,执行如下安装命令:
pip install paho-mqtt
locustfile中首先实现消息发送
这里为了简单起见,使用公共的MQTT broker: EMQX, 它的地址为:"
【Locust如何测试物联网MQTT】broker.emqx.io"
具体实现的代码如下:
broker_add = broker.emqx.io
port = 1883
topic = "/python/mqtt_topic_for_python"
client_id = f"python-mqtt-random.randint(0,100)"
def connect_mqtt():
def on_connect(client,userdata,flags,rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\\n",rc)
client = mqtt_c.Client(client_id)
client.on_connect= on_connect
client.connect(broker_add,port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(3)
msg = f"message: msg_count"
result = client.publish(topic,msg)
status = result[0]
if status == 0:
print(f"send `msg` to topic `topic` ")
else:
print(f"Failed to send message to topic topic")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
完善locustfile中关于Taskset和User的相关配置
具体如下代码,所以会发现,用Python-Locust去测试非HTTP协议的应用系统还是比较方便的,代码即测试。
class TheTaskSet(TaskSet):
@task
def task_1(self):
run()
class TheUser(User):
tasks = [TheTaskSet]
wait_time = constant_pacing(1)
完整的locustfile如下:
import random,time
from paho.mqtt import client as mqtt_c
from locust import TaskSet,task,User,constant_pacing
broker_add = broker.emqx.io
port = 1883
topic = "/python/mqtt_topic_for_python"
client_id = f"python-mqtt-random.randint(0,100)"
def connect_mqtt():
def on_connect(client,userdata,flags,rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\\n",rc)
client = mqtt_c.Client(client_id)
client.on_connect= on_connect
client.connect(broker_add,port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(3)
msg = f"message: msg_count"
result = client.publish(topic,msg)
status = result[0]
if status == 0:
print(f"send `msg` to topic `topic` ")
else:
print(f"Failed to send message to topic topic")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
class TheTaskSet(TaskSet):
@task
def task_1(self):
run()
class TheUser(User):
tasks = [TheTaskSet]
wait_time = constant_pacing(1)
具体的执行结果如下:
Connected to MQTT Broker!
send `message: 0` to topic `/python/mqtt_topic_for_python`
send `message: 1` to topic `/python/mqtt_topic_for_python`
send `message: 2` to topic `/python/mqtt_topic_for_python`
send `message: 3` to topic `/python/mqtt_topic_for_python`
send `message: 4` to topic `/python/mqtt_topic_for_python`
send `message: 5` to topic `/python/mqtt_topic_for_python`
...
参考文档:EMQ官方文档,https://www.emqx.com/zh/blog/how-to-use-mqtt-in-python
推荐阅读
- 老王读Spring IoC-5Spring IoC 小结——控制反转依赖注入
- #yyds干货盘点#JavaSE系列Java程序的封装——Java方法重载及递归
- #yyds干货盘点#Java面试经典100问
- 深入浅出 Java 中枚举的实现原理
- #yyds干货盘点#--docker容器快速入门
- 数仓建设保姆级教程,离线和实时一网打尽(理论+实战)
- xp系统如何转移我的文档保存资料的转移办法
- XP系统如何进行硬盘分区的简便设置
- XP系统不能识别u盘的排查办法的几种原因?