locust（一）http 接口demo

Locust是一款易于使用的分布式用户负载测试工具。它用于对网站（或其他系统）进行负载测试，并确定系统可以处理多少并发用户。
这个想法是，在测试期间，一群蝗虫（Locust）会攻击你的网站。您定义了每个蝗虫Locust（或测试用户）的行为，并且实时地从Web UI监视群集过程。这将有助于您在让真正的用户进入之前进行测试并识别代码中的瓶颈。
Locust完全基于事件，因此可以在一台计算机上支持数千个并发用户。与许多其他基于事件的应用程序相比，它不使用回调。相反，它通过协程（gevent）机制使用轻量级过程。每个蝗虫蜂拥到你的网站实际上是在自己的进程内运行（或者是greenlet，这是正确的）。这允许您在Python中编写非常富有表现力的场景，而不会使代码复杂化。

locust和jemter差不多，单机承受并发量比jmeter大，但是不能设置起始时间。
 

http demo

# -*- coding: utf-8 -*-
# @Time    : 2022/1/20 18:19
# @Author  : tao.cai
import random
from locust import HttpUser,task,SequentialTaskSet,between,TaskSet
from random import choice

class Gi(TaskSet):
   headers = {"Content-Type": "application/json"}

   def on_start(self):
      print("test start...")
   @task()
   def giGreeting(self):
      scene_data = {"record_id": "abc123-zyx890", "from": "914007918", "query": "是的我要寄快递"}
      query = choice(self.user.query_list)
      scene_data.update({"query":query})
      with self.client.post(url='/gi/intent/greeting', headers=self.headers,json= scene_data,
                       catch_response=True) as response:
         if response.status_code == 200:
            response.success()
            print(response.json())
         else:
            response.failure('Failed!')
            
   def on_stop(self):
      print('test over...')
      
class loadUser(HttpUser):
   query_list = ['下单', '查单', '咨询价格', '退回', '修改收方信息']
   wait_time = between(1,2)
   host = 'http://*.*.*.*:50004'
   tasks = [Gi]
if __name__ == "__main__":
   import os
   # os.system("locust -f .\locust_gi.py --host=http://*.*.*.*:50004 --web-host=172.28.0.1") #带页面
   os.system("locust -f .\locust_gi.py --headless -u 10 -r 2 -t 100") #无页面，带时间参数
   #TaskSet 随机选取  SequentialTaskSet 顺序选取

压测结果展示（仅限UI启动）

 展示比较丑陋，也无法持久化,后面会结合Prometheus+grafana做出高大上的压测结果收集