怎么安裝和使用Python輕量級性能工具Locust

這篇文章主要介紹“怎么安裝和使用Python輕量級性能工具Locust”的相關(guān)知識，小編通過實(shí)際案例向大家展示操作過程，操作方法簡單快捷，實(shí)用性強(qiáng)，希望這篇“怎么安裝和使用Python輕量級性能工具Locust”文章能幫助大家解決問題。

專業(yè)成都網(wǎng)站建設(shè)公司，做排名好的好網(wǎng)站，排在同行前面，為您帶來客戶和效益!創(chuàng)新互聯(lián)為您提供成都網(wǎng)站建設(shè)，五站合一網(wǎng)站設(shè)計(jì)制作，服務(wù)好的網(wǎng)站設(shè)計(jì)公司，網(wǎng)站設(shè)計(jì)、做網(wǎng)站負(fù)責(zé)任的成都網(wǎng)站制作公司!

Locust基于python的協(xié)程機(jī)制，打破了線程進(jìn)程的限制，可以能夠在一臺測試機(jī)上跑高并發(fā)

性能測試基礎(chǔ)

1.快慢：衡量系統(tǒng)的處理效率：響應(yīng)時(shí)間

2.多少：衡量系統(tǒng)的處理能力：單位時(shí)間內(nèi)能處理多少個(gè)事務(wù)（tps）

性能測試根據(jù)測試需求最常見的分為下面三類

1 負(fù)載測試load testing

不斷向服務(wù)器加壓，值得預(yù)定的指標(biāo)或者部分系統(tǒng)資源達(dá)到瓶頸，目的是找到系統(tǒng)最大負(fù)載的能力

2 壓力測試

通過高負(fù)載持續(xù)長時(shí)間，來驗(yàn)證系統(tǒng)是否穩(wěn)定

3 并發(fā)測試：

同時(shí)像服務(wù)器提交請求，目的發(fā)現(xiàn)系統(tǒng)是否存在事務(wù)沖突或者鎖升級的現(xiàn)象

性能負(fù)載模型

locust安裝

安裝存在問題，可以通過豆瓣源下載

pip install locust

locust模板

基本上多數(shù)的場景我們都可以基于這個(gè)模板read.py去做修改

from locust import HttpUser, TaskSet, task, tag, events
# 啟動(dòng)locust時(shí)運(yùn)行
@events.test_start.add_listener
def setup(environment, **kwargs):
    # print("task setup")
# 停止locust時(shí)運(yùn)行
@events.test_stop.add_listener
def teardown(environment, **kwargs):
    print("task teardown")
class UserBehavor(TaskSet):
    #虛擬用戶啟用task運(yùn)行
    def on_start(self):
        print("start")
        locusts_spawned.wait()
    #虛擬用戶結(jié)束task運(yùn)行
    def on_stop(self):
        print("stop")
    @tag('test1')
    @task(2)
    def index(self):
        self.client.get('/yetangjian/p/17320268.html')
    @task(1)
    def info(self):
        self.client.get("/yetangjian/p/17253215.html")
class WebsiteUser(HttpUser):
    def setup(self):
        print("locust setup")
    def teardown(self):
        print("locust teardown")
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    min_wait = 3000
    max_wait = 5000

注：這里我們給了一個(gè)webhost，這樣我們可以直接在瀏覽器中打開locust

集合點(diǎn)lr_rendezvous

當(dāng)然我們可以把集合點(diǎn)操作放入上述模板的setup中去運(yùn)行起來

locusts_spawned = Semaphore()
locusts_spawned.acquire()
def on_hatch_complete(**kwargs):
    """
    select_task類的鉤子函數(shù)
    :param kwargs:
    :return:
    """
    locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
n = 0
class UserBehavor(TaskSet):
    def login(self):
        global n
        n += 1
        print(f"第{n}個(gè)用戶登陸")
    def on_start(self):
        self.login()
        locusts_spawned.wait()
    @task
    def test1(self):
        #catch_response獲取返回
        with self.client.get("/yetangjian/p/17253215.html",catch_response=True):
            print("查詢結(jié)束")
class WebsiteUser(HttpUser):
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    wait_time = between(1,3)
if __name__ == '__main__':
    os.system('locust -f read.py --web-host="127.0.0.1"')

比較常見的用法

在上面兩個(gè)例子中我們已經(jīng)看到了一些，例如裝飾器events.test_start.add_listener；events.test_stop.add_listener用來在負(fù)載測試前后進(jìn)行一些操作，又例如on_start、on_stop，在task執(zhí)行前后運(yùn)行，又例如task，可以用來分配任務(wù)的權(quán)重

等待時(shí)間

# wait between 3.0 and 10.5 seconds after each task
#wait_time = between(3.0, 10.5)
#固定時(shí)間等待
# wait_time = constant(3)
#確保每秒運(yùn)行多少次
constant_throughput（task_runs_per_second）
#確保每多少秒運(yùn)行一次
constant_pacing(wait_time)

同樣也可以在User類下發(fā)重寫wait_time來達(dá)到自定義

tag標(biāo)記

@tag('test1')
@task(2)
def index(self):
    self.client.get('/yetangjian/p/17320268.html')

通過對任務(wù)打標(biāo)記，就可以在運(yùn)行時(shí)候執(zhí)行運(yùn)行某一些任務(wù)：

#只執(zhí)行標(biāo)記test1
os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"')
#不執(zhí)行標(biāo)記過的
os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"')
#除去test1執(zhí)行所有
os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')

自定義失敗

#定義響應(yīng)時(shí)間超過0.1就為失敗
with self.client.get("/yetangjian/p/17253215.html", catch_response=True) as response:
    if response.elapsed.total_seconds() > 0.1:
        response.failure("Request took too long")
#定義響應(yīng)碼是200就為失敗
with self.client.get("/yetangjian/p/17320268.html", catch_response=True) as response:
    if response.status_code == 200:
        response.failure("響應(yīng)碼200，但我定義為失敗")

自定義負(fù)載形狀

自定義一個(gè)shape.py通過繼承LoadTestShape并重寫tick

這個(gè)形狀類將以100塊為單位,20速率的增加用戶數(shù)，然后在10分鐘后停止負(fù)載測試（從運(yùn)行開始的第51秒開始user_count會(huì)round到100）

from locust import LoadTestShape
class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20
    def tick(self):
        run_time = self.get_run_time()
        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, self.spawn_rate)
        return None

運(yùn)行圖如下所示

通過命令行去觸發(fā)

os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')

不同時(shí)間階段的例子

from locust import LoadTestShape
class StagesShapeWithCustomUsers(LoadTestShape):
    stages = [
        {"duration": 10, "users": 10, "spawn_rate": 10},
        {"duration": 30, "users": 50, "spawn_rate": 10},
        {"duration": 60, "users": 100, "spawn_rate": 10},
        {"duration": 120, "users": 100, "spawn_rate": 10}]
    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data
        return None

關(guān)于“怎么安裝和使用Python輕量級性能工具Locust”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，小編每天都會(huì)為大家更新不同的知識點(diǎn)。

本文題目：怎么安裝和使用Python輕量級性能工具Locust
本文鏈接：http://bm7419.com/article42/jjcshc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供定制網(wǎng)站、虛擬主機(jī)、小程序開發(fā)、關(guān)鍵詞優(yōu)化、網(wǎng)站維護(hù)、網(wǎng)站策劃

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)