Locust性能測試框架，從入門到精通

1. Locust簡介

Locust是使用Python語言編寫實現的開源性能測試工具，簡潔、輕量、高效，並發機制基於gevent協程，可以實現單機模擬生成較高的並發壓力。

主要特點如下：

1. 使用普通的Python腳本用戶測試場景
2. 分布式和可擴展，支持成千上萬的用戶
3. 基於Web的用戶界面，用戶可以實時監控腳本運行狀態
4. 幾乎可以測試任何系統，除了web http接口外，還可自定義clients測試其他類型系統

2. 安裝

使用pip或easy_install，可以方便安裝Locust

pip install locustio

安裝完成后，可以在shell或cmd中運行locust命令，如查看可用的選項：

locust --help

Locust主要由下面的幾個庫構成：

1. gevent

   gevent是一種基於協程的Python網絡庫，它用到Greenlet提供的，封裝了libevent事件循環的高層同步API。

2. flask

   Python編寫的輕量級Web應用框架。

3. requests

   Python Http庫

4. msgpack-python

   MessagePack是一種快速、緊湊的二進制序列化格式，適用於類似JSON的數據格式。msgpack-python主要提供MessagePack數據序列化及反序列化的方法。

5. six

   Python2和3兼容庫，用來封裝Python2和Python3之間的差異性

6. pyzmq

   pyzmq是zeromq(一種通信隊列)的Python綁定,主要用來實現Locust的分布式模式運行

3. 快速入門

3.1 示例

#coding:utf-8
from locust import HttpLocust, TaskSet, task

class UserBehavior(TaskSet):
    def on_start(self):
        """ on_start is called when a Locust start before any task is scheduled """
        self.login()

    def login(self):
        self.client.post("/login", {"username":"ellen_key", "password":"education"})

    @task(2)
    def index(self):
        self.client.get("/")

    @task(1)
    def profile(self):
        self.client.get("/profile")

class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    host='http://example.com'
    min_wait = 5000
    max_wait = 9000

 

上面是官方的示例demo，定義了針對http://example.com網站的測試場景：先模擬用戶登錄系統，然后隨機地訪問index（/）和profile頁面（/profile/），請求比例為2:1；並且，在測試過程中，兩次請求的間隔時間為5~9秒間的隨機值。

3.2 運行

要運行上述locust腳本，如果文件名為locustfile.py且在當前目錄下，可以這樣運行：

locust

如果locust腳本文件目錄不同或名稱不同，需要使用-f指定文件(--host用來指定測試主機地址)：

locust -f locust_files/my_locust_file.py --host=http://example.com

要運行分布在多個進程上的locust，我們需要使用--master啟動主進程：

locust -f locust_files/my_locust_file.py --master --host=http://example.com

之后使用--slave啟動任意數量的從進程：

locust -f locust_files/my_locust_file.py --slave --host=http://example.com

如果在多台機器上分布式運行locust，我們需要在啟動從進程時指定master-host（默認為127.0.0.1）：

locust -f locust_files/my_locust_file.py --slave --master-host=192.168.0.100 --host=http://example.com

3.3 Locust web模式

Locust默認使用該方式啟動，啟動后在本機打開http://localhost:8089/，可以看到Locust WEB頁面，設置並發用戶數及每秒請求數后即可開始性能測試。

3.4 Locust no-web模式

Locust也可使用no-web模式，使用命令如下：

locust -f locust_files/my_locust_file.py --no-web --csv=locust -c10 -r2 --run-time 1h30m

其中--no-web表示使用no-web模式，--csv表示運行結果文件名，-c 並發用戶數，-r 每秒請求數，--run_time 運行時間

4. locustfile詳解

locustfile中測試場景均是繼承自Locust和TaskSet的子類，下面分別介紹Locust和TaskSet兩個類。

4.1 Locust類

Locust類的client屬性對應虛擬用戶作為客戶端的請求方法。在使用Locust時，需要先繼承Locust類，然后在繼承子類中的client屬性中綁定客戶端的實現類。

對於常見的HTTP(S)協議，Locust已經實現了HttpLocust類，其client屬性綁定了HttpSession類，而HttpSession又繼承自requests.Session。因此在測試HTTP(S)的Locust腳本中，我們可以通過client屬性來使用Python requests庫的所有方法，包括GET/POST/HEAD/PUT/DELETE/PATCH等，調用方式也與requests完全一致。另外，由於requests.Session的使用，因此client的方法調用之間就自動具有了狀態記憶的功能。常見的場景就是，在登錄系統后可以維持登錄狀態的Session，從而后續HTTP請求操作都能帶上登錄狀態。

而對於HTTP(S)以外的協議，我們同樣可以使用Locust進行測試，只是需要我們自行實現客戶端。在客戶端的具體實現上，可通過注冊事件的方式，在請求成功時觸發events.request_success，在請求失敗時觸發events.request_failure即可。然后創建一個繼承自Locust類的類，對其設置一個client屬性並與我們實現的客戶端進行綁定。主要，我們就可以像使用HttpLocust類一樣，測試其它協議類型的系統了。

在Locust類中，除了client屬性，還有幾個屬性需要關注下：

task_set: 指向一個TaskSet類，TaskSet類定義了用戶的任務信息，該屬性為必填；

max_wait/min_wait: 每個用戶執行兩個任務間隔時間的上下限（毫秒），具體數值在上下限中隨機取值，若不指定則默認間隔時間固定為1秒；

host：被測系統的host，當在終端中啟動locust時沒有指定--host參數時才會用到；

weight：同時運行多個Locust類時會用到，用於控制不同類型任務的執行權重。

測試開始后，每個虛擬用戶（Locust實例）的運行邏輯都會遵循如下規律：

1. 先執行WebsiteTasks中的on_start（只執行一次），作為初始化；

2. 從WebsiteTasks中隨機挑選（如果定義了任務間的權重關系，那么就是按照權重關系隨機挑選）一個任務執行；

3. 根據Locust類中min_wait和max_wait定義的間隔時間范圍（如果TaskSet類中也定義了min_wait或者max_wait，以TaskSet中的優先），在時間范圍中隨機取一個值，休眠等待；

4. 重復2~3步驟，直至測試任務終止。

4.2 Taskset類

TaskSet類實現了虛擬用戶所執行任務的調度算法，包括規划任務執行順序（schedule_task）、挑選下一個任務（execute_next_task）、執行任務（execute_task）、休眠等待（wait）、中斷控制（interrupt）等等。在此基礎上，我們就可以在TaskSet子類中采用非常簡潔的方式來描述虛擬用戶的業務測試場景，對虛擬用戶的所有行為（任務）進行組織和描述，並可以對不同任務的權重進行配置。

在TaskSet子類中定義任務信息時，可以采取兩種方式，@task裝飾器和tasks屬性。

采用@task裝飾器定義任務信息時，描述形式如下：

from locust import TaskSet, task
class UserBehavior(TaskSet):
    @task(1)
    def test_job1(self):
        self.client.get('/job1')
    @task(2)
    def test_job2(self):
        self.client.get('/job2')

采用tasks屬性定義任務信息時，描述形式如下：

from locust import TaskSet
def test_job1(obj):
    obj.client.get('/job1')
def test_job2(obj):
    obj.client.get('/job2')
class UserBehavior(TaskSet):
    tasks = {test_job1:1, test_job2:2}
    # tasks = [(test_job1,1), (test_job1,2)] # 兩種方式等價

在如上兩種定義任務信息的方式中，均設置了權重屬性，即執行test_job2的頻率是test_job1的兩倍。

在TaskSet子類中除了定義任務信息，還有一個是經常用到的是on_start函數。這個和LoadRunner中的vuser_init功能相同，在正式執行測試前執行一次，主要用於完成一些初始化的工作。例如，當測試某個搜索功能，而該搜索功能又要求必須為登錄態的時候，就可以先在on_start中進行登錄操作；前面也提到，HttpLocust使用到了requests.Session，因此后續所有任務執行過程中就都具有登錄狀態了。

5. 自定義客戶端測試其他系統

雖然，locust主要是為了測試HTTP而生。然而，它可以很容易地擴展到測試任何基於請求/響應的系統，只需要編寫一個觸發request_success和request_failure事件自定義客戶端即可。

官網提供了詳細的示例，我們簡單修改下就可以用來對任意系統進行性能測試：

import time
from locust import Locust, TaskSet, events, task
import requests


class TestHttpbin(object):
    def status(self):
        try:
            r = requests.get('http://httpbin.org/status/200')
            status_code = r.status_code
            print status_code
            assert status_code == 200, 'Test Index Error: {0}'.format(status_code)
        except Exception as e:
            print e


class CustomClient(object):
    def test_custom(self):
        start_time = time.time()
        try:
            # add your custom test function here
            TestHttpbin().status()
            name = TestHttpbin().status.__name__
        except Exception as e:
            total_time = int((time.time() - start_time) * 1000)
            events.request_failure.fire(request_type="Custom", name=name, response_time=total_time, exception=e)
        else:
            total_time = int((time.time() - start_time) * 1000)
            events.request_success.fire(request_type="Custom", name=name, response_time=total_time, response_length=0)


class CustomLocust(Locust):
    def __init__(self, *args, **kwargs):
        super(CustomLocust, self).__init__(*args, **kwargs)
        self.client = CustomClient()


class ApiUser(CustomLocust):
    min_wait = 100
    max_wait = 1000

    class task_set(TaskSet):
        @task(1)
        def test_custom(self):
            self.client.test_custom()

 

上述腳本里，我們自定義一個測試類TestHttpbin，其中status方法用來校驗接口返回碼；我們只需要在CustomClient類的test_custom方法中添加自定義的測試方法TestHttpbin().status()，即可使用locust對該方法進行負載測試。

taffy框架就是這樣做實現了一份代碼同時進行功能自動化及性能測試。

6. 參考資料

1. https://docs.locust.io/en/latest/
2. http://debugtalk.com/post/head-first-locust-user-guide/