簡介

Locust是一個使用Python編寫的可擴(kuò)展凛忿、分布式的開源性能測試工具澈灼。

優(yōu)點

• 相比于Jmeter、LoadRunner這種基于GUI的工具而言店溢，Locust使用Python語言來描述測試場景使模擬用戶行為變得更加靈活和簡潔叁熔，除了Http(s)協(xié)議之外，Locust可以測試任意協(xié)議的系統(tǒng)床牧，只需要實現(xiàn)Python調(diào)用對應(yīng)協(xié)議的庫進(jìn)行請求即可(類似HttpLocust類)荣回。
• Locust的并發(fā)機(jī)制采用協(xié)程的方式，相比于進(jìn)程和線程減少了系統(tǒng)級資源調(diào)度戈咳，因此單機(jī)的產(chǎn)生的并發(fā)能力相比于LoadRunner心软、jmeter得到了大幅的提升。

安裝

pip install locust

Locust腳本編寫

import queue
from locust import HttpLocust, TaskSet, task
from locust.clients import HttpSession
from sign import ParserData


class UserBehavior(TaskSet):

    parser_data = ParserData()  # 解析接口傳參類
    _client = 20
    version = 119
    user_info = None

    @staticmethod
    def get_user_info(response):
        r = response.json().get('content')
        return {
            'market_id': r.get('marketId'),
            'token': r.get('token')
        }

    def on_start(self):
        try:
            user, password = self.locust.users.get_nowait()
        except queue.Empty:
            print('test data run out. test ended.')
            exit(0)
        client = HttpSession(base_url='http://login.xxxx.cn')
        data = self.parser_data(loginName=user,
                                password=password,
                                client=self._client,
                                version=self.version)
        response = client.post(url='/login', data=data)
        self.user_info = self.get_user_info(response)
        self.locust.users.put_nowait((user, password))

    @task(2)
    def index(self):
        url = '/index'
        data = self.parser_data(market_id=self.user_info['market_id'],
                                client=self._client,
                                version=self.version,
                                pnum='3')
        headers = {'Authorization': 'Barer:' + self.user_info['token'],
                   'Accept': 'application/vnd.hs-api.v1+json'}
        with self.client.post(url=url, data=data, headers=headers,
                              verify=False, catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure('http error.')

    @task(1)
    def shop_car_list(self):
        url = '/shopCar/list'
        data = self.parser_data(market_id=self.user_info['market_id'],
                                ischaidan='1',
                                client=self._client,
                                version=self.version)
        headers = {'Authorization': 'Bearer:' + self.user_info['token'],
                   'Accept': 'application/vnd.hs-api.v1+json'}
        with self.client.post(name='ShopCar', url=url, data=data, headers=headers,
                              verify=False, catch_response=True) as response:
            if response.status_code != 200 or "失敗" in response.text:
                response.failure('response error.')
            else:
                response.success()


class Stay(TaskSet):

    index = 0

    def on_start(self):
        self.index += 1

    @task
    def get_error(self):
        response = self.client.get('/1', name='error', allow_redirects=False,
                                   verify=False, catch_response=True)
        if response.status_code == 200:
            response.success()
        else:
            response.failure('http error.')

    @task
    def logout(self):
        self.interrupt()


class User(TaskSet):

    tasks = {Stay: 1}

    @task(1)
    def user(self):
        self.client.get('/', verify=False)


class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    host = 'https://xxxx.api.xxxx.cn'
    min_wait = 1000
    max_wait = 3000
    users = queue.Queue()
    users.put_nowait(('user1', '1232'))
    users.put_nowait(('user2', '1234'))
    users.put_nowait(('user3', '1321'))
    weight = 3
    stop_timeout = 20


class WebsiteU(HttpLocust):
    task_set = User
    host = 'https://www.baidu.com'
    min_wait = 0
    max_wait = 0
    weight = 1
    stop_timeout = 60

簡單解釋下：

• UserBehavior和WebsiteUser兩個類實現(xiàn)測試場景使用3個用戶賬號著蛙，每個用戶會去先登錄删铃，然后分別去查看首頁和進(jìn)入購物車頁面
• 首先導(dǎo)入了需要用到的類，HttpLocust類為Locust子類踏堡，模擬客戶端的請求類猎唁，Taskset類為任務(wù)集類，task為任務(wù)裝飾器顷蟆。

Taskset

• UserBehavior為Taskset子類诫隅，該類主要用來定義每個虛擬用戶的操作行為

• Taskset子類中可以定義一個on_start方法在正式開始測試前只執(zhí)行一次，相當(dāng)與初始化操作(這里說的只執(zhí)行一次是每個虛擬用戶都會去執(zhí)行一次)帐偎，比如獲取登錄token等操作逐纬。

• teskset類中的每個任務(wù)都需要用@task(weight=1)裝飾器去裝飾為一個任務(wù)，weight為執(zhí)行的權(quán)重削樊，如果不裝飾豁生，locust不會認(rèn)為這是一個任務(wù)，UserBehavior類中@task(1)、@task(2)裝飾器表示3個用戶里面有2個用戶去模擬執(zhí)行index方法沛硅，有1個用戶去執(zhí)行shop_car_list方法

• taskset類中self.client屬性請求操作時傳入catch_response參數(shù)眼刃，設(shè)置為True，可以標(biāo)記響應(yīng)結(jié)果為成功或失敗摇肌，即使響應(yīng)是成功的擂红，也可以標(biāo)記為失敗，默認(rèn)為Fasle

• taskset類中self.client屬性請求操作時有一個name參數(shù)围小，當(dāng)設(shè)置了name的值時昵骤，最后的請求結(jié)果展示中name字段會顯示這里定義的name值，相當(dāng)與給這個方法起了一個別名.

• taskset類中interrupt(reschedule=True)方法在頂層的taskset類(即被指定到Locust子類中的taskset)中不可用肯适，reschedule為True時变秦，從被嵌套的任務(wù)中出來立即執(zhí)行新的任務(wù)，如果為False從被嵌套的任務(wù)中出來會等待min_time-max_time之間的隨機(jī)時間框舔，然后再執(zhí)行新的任務(wù)蹦玫，這個方法主要用來跳出嵌套的任務(wù)集

HttpLocust

• WebsiteUser為HttpLocust子類，該類是用來模擬用戶的類刘绣，定義了一些用戶信息樱溉，及請求方式

• HttpLocust子類中task_set屬性用來指定模擬用戶執(zhí)行的操作，即Taskset子類

• HttpLocust子類中的host屬性為被測試系統(tǒng)的host纬凤，當(dāng)命令行中沒有指定--host參數(shù)時福贞，此屬性會生效

• HttpLocust子類中的min_wait、max_weight為最大等待時間和最小等待時間停士，每個請求會從這兩個時間間隔中隨機(jī)取一個時間等待挖帘，相當(dāng)用戶實際操作系統(tǒng)時每個動作的思考時間。若測試單個接口則對應(yīng)的值都設(shè)置為0即可恋技。如果Taskset類中定義了min_wait拇舀、max_weight則會覆蓋Locust子類中定義的值。單位ms蜻底，默認(rèn)值1000ms

• HttpLocust子類中的stop_timeout屬性為執(zhí)行測試的時間骄崩，單位為s

• HttpLocust子類中的weight為該類執(zhí)行的權(quán)重，當(dāng)有多個子類時生效朱躺，如WebsiteUser刁赖、WebsiteU兩個HttpLocust子類中weight值分別為3和1.

• Locust默認(rèn)單機(jī)單進(jìn)程運行搁痛，此模式下并不能充分利用單機(jī)的多處理器长搀，可使用分布式運行，即開啟一個master鸡典，n個slave(n為處理器個數(shù))源请，master負(fù)責(zé)啟動Locust的web服務(wù)和任務(wù)分發(fā)，不會產(chǎn)生壓力，slave主要負(fù)責(zé)產(chǎn)生壓力

運行模式：

no-web模式

no_web模式指在命令行中直接運行

locust -f load_test.py -c 1 -r 1 -n 1

• -f 指定要運行的Locust性能測試文件
• -c 指定模擬的并發(fā)用戶數(shù)
• -r 指定每秒的啟動用戶數(shù)
• -n 指定運行次數(shù)
• -t 指定運行的時間谁尸，例如300s,1m,1h
  寫完腳本調(diào)試時可在該模式下運行

單機(jī)單進(jìn)程運行

locust -f load_test.py

分布式運行

分布式運行舅踪，有單機(jī)多進(jìn)程運行和多機(jī)多進(jìn)程運行兩種

locust -f load_test.py --master       master模式下啟動locust
locust -f load_test.py --slave         啟動一個locust slave節(jié)點，單機(jī)多進(jìn)程模式
locust -f load_test.py --slave --master-host=192.168.105.11    啟動一個locust slave節(jié)點良蛮，多機(jī)模式下 

no_web模式下運行

image.png

web模式

image.png

• Number of users to simulate：需要模擬的虛擬用戶個數(shù)

• Hatch rate (users spawned/second)：啟動虛擬用戶的速率抽碌，每秒產(chǎn)生出多少個用戶數(shù)

  
  
  QQ圖片20171109190334.png
• 顯示并發(fā)數(shù)、響應(yīng)時間决瞳、異常率货徙、每秒請求數(shù)等

• reqs/sec(每秒請求數(shù))為根據(jù)最近2s請求數(shù)據(jù)計算得到的數(shù)據(jù)，即瞬時值

  
  
  QQ圖片20171109190430.png

• 顯示rps皮胡、響應(yīng)時間痴颊、并發(fā)數(shù)在整個測試運行中的走勢圖

  
  
  QQ圖片20171109190455.png
• 顯示測試過程中出現(xiàn)的所有失敗的請求

Exceptions顯示測試過程中出現(xiàn)的異常
Download Date提供測試結(jié)果csv文件的下載

測試數(shù)據(jù)：

• locust子類中設(shè)置的數(shù)據(jù)是全局的，為list時屡贺，使用自增方式取數(shù)據(jù)蠢棱，并發(fā)運行時會出現(xiàn)取到的數(shù)據(jù)重復(fù)的情況，如果對數(shù)據(jù)唯一性有要求甩栈，使用python的queue隊列的數(shù)據(jù)結(jié)構(gòu)即可
• taskset子類中設(shè)置的數(shù)據(jù)是局部的泻仙，即每一個虛擬用戶都會有一個屬于自己的這個變量

queue數(shù)據(jù)結(jié)構(gòu)

隊列形式

q=queue.Queue(maxsize=3)       先進(jìn)先出隊列
q.put(1)      向隊列中存數(shù)據(jù)
q.get()        向隊列中取數(shù)據(jù)
q.put_nowait()       相當(dāng)于q.put(1, block=False)，當(dāng)q隊列滿了之后put會觸發(fā)queue.Full異常
q.get_nowait()       相當(dāng)與q.get(block=False),當(dāng)q隊列為空之后get會觸發(fā)queue.Empty異常