做网站的公司有多少家,建设网站5要素,一般注册公司要多少钱,公司简介模板下载一、背景简介
业务背景大概介绍一下#xff0c;就是按照国标规定#xff0c;车辆需要上传一些指定的数据到ZF的指定平台#xff0c;同时车辆也会把数据传到企业云端服务上#xff0c;于是乎就产生了一些性能需求。
目前我们只是先简单的进行了一个性能场景的测试#xf…一、背景简介
业务背景大概介绍一下就是按照国标规定车辆需要上传一些指定的数据到ZF的指定平台同时车辆也会把数据传到企业云端服务上于是乎就产生了一些性能需求。
目前我们只是先简单的进行了一个性能场景的测试就是评估目前服务是否能够支持预期的最大同时在线车辆上传数据。经过评估在线车辆数据按照预期的10倍来进行的并且后面增加持续运行12h查看服务链路的稳定性。
本篇并不是一个严谨的性能测试过程结果分享主要是分享下关于mqtt协议服务的压测脚本的编写。因为之前我也没接触过MQTT协议的压测网上关于相关的压测脚本的内容也比较杂乱所以记录一下仅供参考。
捋一下链路就知道需要生成哪些数据因为服务还未上线使用所以产生的压测数据后面可以直接清理掉即可。
一些前置数据比如数据库、缓存里涉及到的车辆数据通信秘钥数据等等这些可以之前写脚本一次性生成即可。车辆上报的数据车辆上报到云端的数据是经过一系列加密转码期间还要设计到解密等这个经过评估可以简化其中的某些环境所以所有的车可以直接发送相同的数据即可。车辆数据最后就是生成对应的车辆数据同时在线按照评估的频率发送数据。
其中第1、2的数据在之前针对性的分别生成即可第3步的车辆发送数据就是压测脚本要干的事情了。
二、技术选型
这个倒是很快搜索引擎大概搜了一下内容很少或者说对我有用的内容很少。有看到jmeter有相关插件的但是这个方案基本上我都是否决的一来我不擅长用而来我觉得用起来肯定会比自己编码要麻烦的多。
所以就继续编码好了仍然首选python想到了locust库后来看官方文档的时候看到locust也针对mqtt协议拓展了一些内容。但是我尝试下来不太符合我这的需求也可能当时我用的不对吧所以就只能自己来从零开始编写了。
搜索中又发现Python中用于mqtt协议的库叫paho.mqtt支持连接代理消息的订阅、收发等等于是最后确定使用locustpaho.mqtt的组合来实现本次的负载脚本。
三、代码编写
1. 脚本代码
暂时没做代码分层目前场景简单就直接都放一个模块里了有点长先贴上来后面部分会对脚本的重点内容进行拆解。
脚本目前做了这些事情
从db中查询有效可用的所有测试车辆信息数据根据命令行的输入参数指定启动的车辆数以及与broker代理建立连接的频率建立连接成功的车辆就可以根据脚本里指定的频次来像broker发送数据脚本统计连接数、请求数、响应时间等信息写到报表中调试遇到车辆会批量断开连接的情况增加了当车辆断开连接时把断开时间、车辆信息写到本地csv中方便第二天来查看分析。
import csv
import datetime
import queue
import os
import sys
import time
import sslfrom 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 DBfrom locust import User, TaskSet, events, task, between, run_single_userBROKER_ADDRESS broker服务地址
PORT 1111
PASSWORD 111111
PUBLISH_TIMEOUT 10000 # 超时时间
TEST_TOPIC test_topicTEST_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 topicself.payload payloadself.start_time start_timeself.timeout timeout# 统计总共发送成功的消息数量
total_published 0
disconnect_record_list [] # 定义存放连接断开的记录的列表容器class PublishTask(TaskSet):taskdef task_publish(self):self.client.loop_start()topic TEST_TOPICpayload BYTES_DATA# 记录发送的开始时间start_time time.time()mqtt_msg_info self.client.publish(topic, payload, qos1, retainFalse)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_disconnectstaticmethoddef 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)))staticmethoddef 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_typep_success,nameclient_id: str(client._client_id),response_timepublish_resp_time,response_lengthlen(message.payload),exceptionNone,contextNone)global total_published# 成功发送累加1total_published 1class MQTTLocustUser(User):tasks [PublishTask]wait_time between(2, 2)def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)# 从队列中获取客户端 username 和 client_idcurrent_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):# 设置tlscontext ssl.SSLContext(ssl.PROTOCOL_TLS)self.client.tls_set_context(context)self.client.connect(hostBROKER_ADDRESS, portPORT, keepalive60)self.client.on_connect self.on_connectdef on_stop(self):print(publish 成功, 当前已成功发送数量{}.format(total_published))if len(disconnect_record_list) 0:print(无断开连接的client)else:# 把断开记录里的信息写入csvwith open(disconnect_record.csv, w, newline, encodingUTF8) 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文件)staticmethoddef on_connect(client, userdata, flags, rc, propsNone):if rc 0:print(rc状态:{} - -.format(rc), {}-连接broker成功.format(str(client._client_id)))fire_success(request_typec_success,namecount_connected,response_time0,response_length0,exceptionNone,contextNone)else:print(rc状态:{} - -.format(rc), {}-连接broker失败.format(str(client._client_id)))fire_success(request_typec_fail,nameclient_id: str(client._client_id),response_time0,response_length0,exceptionNone,contextNone)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():声明一个clientclient.username_pw_set(): 设置客户端的用户名密码client.tls_set_context 设置ssl模式client.connect(): 连接代理client.publish:向代理推送消息
还用到了一些回调函数
on_connect连接操作成功时回调on_publish发布成功时回调on_disconnect客户端与代理断开连接时回调
另外还用到了一个事件函数events.request。 当客户端发送请求时会调用不管是请求成功还是请求失败当我需要自定义我的报告内容时就需要用到这个event。 查看源码知道里面要传哪些参数那我们在调用时候就需要传入对应的参数。
比如我在发送回调函数里调用了该方法。 所以最后在控制台显示的报告里就有我定义的内容了。 由于后来在使用中发现不知道会在什么时候出现批量断开的情况于是在on_disconnect回调函数里增加了对应处理把相关的断开信息记录下来运行结束的时候写到本地文件里去。 后来我主动尝试客户端断开的情况测试了下文件的写入结果功能正常。 三、小结
后面就开始运行了在运行过程中开发关注链路服务的各项指标这里就不展开了业务缠身就并没有过多的去做这个事情况且也不专业。确实也发现了不少问题后面逐步优化再继续测试。
现在稳定运行12h服务正常暂时就先告一段落了。后面还有会相关其他性能测试场景届时就可以针对性的展开分享下了。
另外这个脚本分享也只是仅供参考现在我这是使用简单本着能用就行可能存在一些不合理需要优化的地方有需要的朋友还请自行查阅相关文档。
