日韩性视频-久久久蜜桃-www中文字幕-在线中文字幕av-亚洲欧美一区二区三区四区-撸久久-香蕉视频一区-久久无码精品丰满人妻-国产高潮av-激情福利社-日韩av网址大全-国产精品久久999-日本五十路在线-性欧美在线-久久99精品波多结衣一区-男女午夜免费视频-黑人极品ⅴideos精品欧美棵-人人妻人人澡人人爽精品欧美一区-日韩一区在线看-欧美a级在线免费观看

歡迎訪問 生活随笔!

生活随笔

當前位置: 首頁 > 编程资源 > 编程问答 >内容正文

编程问答

locust压测mqtt

發布時間:2024/3/24 编程问答 55 豆豆
生活随笔 收集整理的這篇文章主要介紹了 locust压测mqtt 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Locust與Jmeter對比:

發壓能力:相同并發下,Locust(使用FastHttpLocust)> Jmeter

并發能力:Locust和Jmeter旗鼓相當,都能滿足工作需求,Jmeter消耗的內存更高

結果報表:Jmeter好于Locust,但是基本都滿足工作需求

學習成果:Jmeter>Locust

易用性:Jmeter > Locust

locust可以實現模擬大批量用戶,并且跟蹤每個用戶的請求數量、失敗數、平均響應時間,jmeter就沒有這個效果,只是一個匯總的測試結果,并且locust適合mqtt這種千、萬級請求,jmeter性能就沒有這么強
下面的代碼默認已經安裝好相關依賴

import paho.mqtt.client as mqtt import time import random from locust import TaskSet, task, User, constant_pacing, between, eventsHOST = "127.0.0.1" # 服務器ip PORT = 15675 COUNTClient = 0def fire_locust_success(**kwargs):events.request_success.fire(**kwargs)def increment():global COUNTClientCOUNTClient = COUNTClient+1def time_delta(t1, t2):return int((t2 - t1)*1000)def on_message(client, userdata, msg):print(msg.topic+" "+msg.payload.decode("utf-8"))# 消息處理def client_loop():start_time = time.time()# client_id = str(int(time.time()) + COUNTClient)client_id = str(1000 + COUNTClient)client = mqtt.Client(client_id, transport="websockets") # ClientId不能重復;協議為websocketsclient.username_pw_set("username", "password") # 賬號密碼,一般都需要設置client.ws_set_options("/ws") # 協議路徑res = client.connect(HOST, PORT, 10) # 建立連接increment()name = "device" + str(COUNTClient)le = len(str(res))def on_connect(client, userdata, flags, rc):"""改寫locust中的返回,不然壓測界面沒有返回數據"""end_time = time.time()if rc == 0:events.request_success.fire(request_type='MQTT',name='name',response_time=time_delta(start_time, end_time),response_length=le)else:events.request_failure.fire(request_type='MQTT',name=name,response_time=time_delta(start_time, end_time),response_length=le,exception=rc)client.on_connect = on_connectclient.on_message = on_messageres2 = client.subscribe("/USER20220815/ADMIN/WEB/MC/NOTICE", qos=0)print("Response json:", str(client_id)+str(res))print(str(22222) + str(res2) + str(3333))client.loop_forever()class TheTaskSet(TaskSet):def on_start(self):print("開始的時候執行")def on_stop(self):print("結束的時候執行")@taskdef task_1(self):client_loop()class TheUser(User):tasks = [TheTaskSet]# wait_time = between(1, 2)wait_time = constant_pacing(1)

總結

以上是生活随笔為你收集整理的locust压测mqtt的全部內容,希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯,歡迎將生活随笔推薦給好友。