由於人工智能的熱度， python目前已經成為最受歡迎的編程語言，一度已經超越Java 。
本文將介紹開源的python 測試工具： locust
使用步驟：
1. 安裝python 3.0以上版本
2. 安裝Pip
3. 安裝locust pip install locustio (windows系統下)
4. 閱讀或者下載 locust 源碼
一、Locust 的基本實現原理
服務端性能測試工具最核心的部分是壓力發生器，核心要點有兩個，一是真實模擬用戶操作，二是模擬有效並發。
在Locust測試框架中，測試場景是采用純Python腳本。對於最常見的HTTP(S)協議的系統，Locust采用Python的requests庫作為客戶端，而對於其它協議類型的系統，Locust也提供了接口，只要我們能采用Python編寫對應的請求客戶端，就能方便地采用Locust實現壓力測試。從這個角度來說，Locust可以用於壓測任意類型的系統。
在模擬有效並發方面，Locust的優勢在於其摒棄了進程和線程，完全基於事件驅動，使用gevent提供的非阻塞IO和coroutine來實現網絡層的並發請求，因此即使是單台壓力機也能產生數千並發請求數；再加上對分布式運行的支持，理論上來說，Locust能在使用較少壓力機的前提下支持極高並發數的測試。
二、 Locust 腳本編寫
首先分析下官方demo腳本：
import random from locust import HttpLocust, TaskSet, task from pyquery import PyQuery class BrowseDocumentation(TaskSet): def on_start(self): # assume all users arrive at the index page self.index_page() self.urls_on_current_page = self.toc_urls @task(10) def index_page(self): r = self.client.get("/") pq = PyQuery(r.content) link_elements = pq(".toctree-wrapper a.internal") self.toc_urls = [ l.attrib["href"] for l in link_elements ] @task(50) def load_page(self, url=None): url = random.choice(self.toc_urls) r = self.client.get(url) pq = PyQuery(r.content) link_elements = pq("a.internal") self.urls_on_current_page = [ l.attrib["href"] for l in link_elements ] @task(30) def load_sub_page(self): url = random.choice(self.urls_on_current_page) r = self.client.get(url) class AwesomeUser(HttpLocust): task_set = BrowseDocumentation host = "http://docs.locust.io/en/latest/" # we assume someone who is browsing the Locust docs, # generally has a quite long waiting time (between # 20 and 600 seconds), since there's a bunch of text # on each page min_wait = 20 * 1000 max_wait = 600 * 1000
在這個示例中，定義了針對host=http://docs.locust.io/en/latest/ 網站的測試場景：先模擬用戶登錄系統，然后隨機地訪問首頁（/）和關於頁面（/about/），請求比例為2:1；並且，在測試過程中，兩次請求的間隔時間為20~600秒間的隨機值。
那么，如上Python腳本是如何表達出以上測試場景的呢？
從腳本中可以看出，腳本主要包含兩個類，一個是WebsiteUser（繼承自HttpLocust，而HttpLocust繼承自Locust），另一個是WebsiteTasks（繼承自TaskSet）。事實上，在Locust的測試腳本中，所有業務測試場景都是在Locust和TaskSet兩個類的繼承子類中進行描述的。
Locust類
簡單地說，Locust類就好比是一群蝗蟲，而每一只蝗蟲就是一個類的實例。
相應的，TaskSet類就好比是蝗蟲的大腦，控制着蝗蟲的具體行為，即實際業務場景測試對應的任務集。
在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實例）的運行邏輯都會遵循如下規律：
- 先執行
WebsiteTasks中的on_start（只執行一次），作為初始化； - 從
WebsiteTasks中隨機挑選（如果定義了任務間的權重關系，那么就是按照權重關系隨機挑選）一個任務執行； - 根據
Locust類中min_wait和max_wait定義的間隔時間范圍（如果TaskSet類中也定義了min_wait或者max_wait，以TaskSet中的優先），在時間范圍中隨機取一個值，休眠等待； - 重復
2~3步驟，直至測試任務終止。
TaskSet類
性能測試工具要模擬用戶的業務操作，就需要通過腳本模擬用戶的行為。在前面的比喻中說到，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)] # 兩種方式等價
Locust 用例高級用法
關聯
在某些請求中，需要攜帶之前從Server端返回的參數，因此在構造請求時需要先從之前的Response中提取出所需的參數。
from lxml import etree from locust import TaskSet, task, HttpLocust class UserBehavior(TaskSet): @staticmethod def get_session(html): tree = etree.HTML(html) return tree.xpath("//div[@class='btnbox']/input[@name='session']/@value")[0] @task(10) def test_login(self): html = self.client.get('/login').text username = 'user@compay.com' password = '123456' session = self.get_session(html) payload = { 'username': username, 'password': password, 'session': session } self.client.post('/login', data=payload) class WebsiteUser(HttpLocust): host = 'http://debugtalk.com' task_set = UserBehavior min_wait = 1000 max_wait = 3000
參數化
循環取數據，數據可重復使用
所有並發虛擬用戶共享同一份測試數據，各虛擬用戶在數據列表中循環取值。
例如，模擬3用戶並發請求網頁，總共有100個URL地址，每個虛擬用戶都會依次循環加載這100個URL地址；加載示例如下表所示。
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
from locust import TaskSet, task, HttpLocust class UserBehavior(TaskSet): def on_start(self): self.index = 0 @task def test_visit(self): url = self.locust.share_data[self.index] print('visit url: %s' % url) self.index = (self.index + 1) % len(self.locust.share_data) self.client.get(url) class WebsiteUser(HttpLocust): host = 'http://debugtalk.com' task_set = UserBehavior share_data = ['url1', 'url2', 'url3', 'url4', 'url5'] min_wait = 1000 max_wait = 3000
保證並發測試數據唯一性，不循環取數據
所有並發虛擬用戶共享同一份測試數據，並且保證虛擬用戶使用的數據不重復。
例如，模擬3用戶並發注冊賬號，總共有9個賬號，要求注冊賬號不重復，注冊完畢后結束測試；加載示例如下表所示。
from locust import TaskSet, task, HttpLocust import queue class UserBehavior(TaskSet): @task def test_register(self): try: data = self.locust.user_data_queue.get() except queue.Empty: print('account data run out, test ended.') exit(0) print('register with user: {}, pwd: {}'\ .format(data['username'], data['password'])) payload = { 'username': data['username'], 'password': data['password'] } self.client.post('/register', data=payload) class WebsiteUser(HttpLocust): host = 'http://debugtalk.com' task_set = UserBehavior user_data_queue = queue.Queue() for index in range(100): data = { "username": "test%04d" % index, "password": "pwd%04d" % index, "email": "test%04d@debugtalk.test" % index, "phone": "186%08d" % index, } user_data_queue.put_nowait(data) min_wait = 1000 max_wait = 3000
保證並發測試數據唯一性，循環取數據
所有並發虛擬用戶共享同一份測試數據，保證並發虛擬用戶使用的數據不重復，並且數據可循環重復使用。
例如，模擬3用戶並發登錄賬號，總共有9個賬號，要求並發登錄賬號不相同，但數據可循環使用；加載示例如下表所示。
from locust import TaskSet, task, HttpLocust import queue class UserBehavior(TaskSet): @task def test_register(self): try: data = self.locust.user_data_queue.get() except queue.Empty: print('account data run out, test ended.') exit(0) print('register with user: {}, pwd: {}'\ .format(data['username'], data['password'])) payload = { 'username': data['username'], 'password': data['password'] } self.client.post('/register', data=payload) self.locust.user_data_queue.put_nowait(data) class WebsiteUser(HttpLocust): host = 'http://debugtalk.com' task_set = UserBehavior user_data_queue = queue.Queue() for index in range(100): data = { "username": "test%04d" % index, "password": "pwd%04d" % index, "email": "test%04d@debugtalk.test" % index, "phone": "186%08d" % index, } user_data_queue.put_nowait(data) min_wait = 1000 max_wait = 3000
三、Locust運行模式
運行Locust時，通常會使用到兩種運行模式：單進程運行和多進程分布式運行。
單進程運行模式
Locust所有的虛擬並發用戶均運行在單個Python進程中，具體從使用形式上，又分為no_web和web兩種形式。該種模式由於單進程的原因，並不能完全發揮壓力機所有處理器的能力，因此主要用於調試腳本和小並發壓測的情況。
當並發壓力要求較高時，就需要用到Locust的多進程分布式運行模式。從字面意思上看，大家可能第一反應就是多台壓力機同時運行，每台壓力機分擔負載一部分的壓力生成。的確，Locust支持任意多台壓力機（一主多從）的分布式運行模式，但這里說到的多進程分布式運行模式還有另外一種情況，就是在同一台壓力機上開啟多個slave的情況。這是因為當前階段大多數計算機的CPU都是多處理器（multiple processor cores），單進程運行模式下只能用到一個處理器的能力，而通過在一台壓力機上運行多個slave，就能調用多個處理器的能力了。比較好的做法是，如果一台壓力機有N個處理器內核，那么就在這台壓力機上啟動一個master，N個slave。當然，我們也可以啟動N的倍數個slave，但是根據我的試驗數據，效果跟N個差不多，因此只需要啟動N個slave即可。
Locust是通過在Terminal中執行命令進行啟動的，通用的參數有如下幾個：
-H, --host：被測系統的host，若在Terminal中不進行指定，就需要在Locust子類中通過host參數進行指定；--no-web參數，指定並發數（-c）和總執行次數（-n）-f, --locustfile：指定執行的Locust腳本文件；
在此基礎上，當我們想要調試Locust腳本時，就可以在腳本中需要調試的地方通過print打印日志，然后將並發數和總執行次數都指定為1
$ locust -f locustfile.py --no-web -c 1 -n 1 no_web
如果采用no_web形式，則需使用--no-web參數，並會用到如下幾個參數。
-c, --clients：指定並發用戶數；-n, --num-request：指定總執行測試次數；-r, --hatch-rate：指定並發加壓速率，默認值位1。
示例：
$ locust -H http://debugtalk.com -f demo.py --no-web -c 1 -n 2
web
如果采用web形式，，則通常情況下無需指定其它額外參數，Locust默認采用8089端口啟動web；如果要使用其它端口，就可以使用如下參數進行指定。
-P, --port：指定web端口，默認為8089.
$ locust -H http://XXXX.com -f demo.py如果Locust運行在本機，在瀏覽器中訪問http://localhost:8089即可進入Locust的Web管理頁面；如果Locust運行在其它機器上，那么在瀏覽器中訪問http://locust_machine_ip:8089即可。
在Locust的Web管理頁面中，需要配置的參數只有兩個：
Number of users to simulate: 設置並發用戶數，對應中no_web模式的-c, --clients參數；Hatch rate (users spawned/second): 啟動虛擬用戶的速率，對應着no_web模式的-r, --hatch-rate參數，默認為1。
多進程分布式運行
不管是單機多進程，還是多機負載模式，運行方式都是一樣的，都是先運行一個master，再啟動多個slave。
啟動master時，需要使用--master參數；同樣的，如果要使用8089以外的端口，還需要使用-P, --port參數。
$ locust -H http://xxxx.com -f demo.py --master --port=8088 master啟動后，還需要啟動slave才能執行測試任務。
啟動slave時需要使用--slave參數；在slave中，就不需要再指定端口了。
$ locust -H http://xxxx.com -f demo.py --slave 如果slave與master不在同一台機器上，還需要通過--master-host參數再指定master的IP地址。
$ locust -H http://xxxx.com -f demo.py --slave --master-host=<locust_machine_ip> master和slave都啟動完畢后，就可以在瀏覽器中通過http://locust_machine_ip:8089進入Locust的Web管理頁面了。使用方式跟單進程web形式完全相同，只是此時是通過多進程負載來生成並發壓力，在web管理界面中也能看到實際的slave數量。
注意：
locust雖然使用方便，但是加壓性能和響應時間上面還是有差距的，如果項目有非常大的並發加壓請求，可以選擇wrk
對比方法與結果：
可以准備兩台服務器，服務器A作為施壓方，服務器B作為承壓方
服務器B上簡單的運行一個nginx服務就行了
服務器A上可以安裝一些常用的壓測工具，比如locust、ab、wrk
我當時測下來，施壓能力上 wrk > golang >> ab > locust
因為locust一個進程只使用一核CPU，所以用locust壓測時，必須使用主從分布式（zeromq通訊）模式，並根據服務器CPU核數來起slave節點數
wrk約為55K QPS
golang net/http 約 45K QPS
ab 大約 15K QPS
locust 最差，而且response time明顯比較長