前言
- 有时候我们需要一个完全定制的负载测试，而这并不能通过简单地设置或改变用户数量和刷出率而实现。例如，可能希望在自定义时间生成一个负载尖峰或上升或下降。通过使用LoadTestShape类，您可以在任何时候完全控制用户计数和生成速率。
基于时间峰值策略
- 每秒生成10个用户，持续时间60s
1. # -*- coding: utf-8 -*- 2. # @Time : 2021/5/2 3. # @Author : 大海 4. 5. import os 6. from locust import LoadTestShape, task, HttpUser, constant 7. 8. 9. class MyUser(HttpUser): 10. # 请求间隔时间 11. wait_time = constant(1) 12. host = 'https://www.baidu.com' 13. 14. @task 15. def my_task(self): 16. self.client.get('/') 17. 18. 19. class MyCustomShape(LoadTestShape): 20. # 设置压测持续时长，单位秒 21. time_limit = 60 22. # 每秒启动/停止用户数 23. spawn_rate = 10 24. 25. def tick(self): 26. """ 27. 返回一个元组，包含两值： 28. user_count -- 总用户数 29. spawn_rate -- 每秒启动/停止用户数 30. 返回None时，停止负载测试 31. """ 32. # 获取压测执行的时间 33. run_time = self.get_run_time() 34. 35. # 运行时长在压测最大时长内，则继续执行 36. if run_time < self.time_limit: 37. user_count = round(run_time, -1) 38. return user_count, self.spawn_rate 39. return None 40. 41. 42. if __name__ == '__main__': 43. file_path = os.path.abspath(__file__) 44. os.system(f'locust -f {file_path} --web-host=127.0.0.1')
基于步骤负载策略
- 每30秒增加10个用户，持续10分钟
1. # -*- coding: utf-8 -*- 2. # @Time : 2021/5/2 3. # @Author : 大海 4. import math 5. import os 6. from locust import HttpUser, TaskSet, task, constant 7. from locust import LoadTestShape 8. 9. 10. class UserTasks(TaskSet): 11. @task 12. def my_task(self): 13. self.client.get("/") 14. 15. 16. class WebsiteUser(HttpUser): 17. wait_time = constant(0.5) 18. tasks = [UserTasks] 19. host = 'https://www.baidu.com' 20. 21. 22. class StepShape(LoadTestShape): 23. """ 24. step_time -- 步骤之间的时间 25. step_load -- 用户在每一步增加数量 26. spawn_rate -- 用户在每一步式每秒停止/启动数量 27. time_limit -- 时间限制，以秒为单位 28. """ 29. 30. step_time = 30 31. step_load = 10 32. spawn_rate = 10 33. time_limit = 600 34. 35. def tick(self): 36. run_time = self.get_run_time() 37. 38. if run_time > self.time_limit: 39. return None 40. 41. current_step = math.floor(run_time / self.step_time) + 1 42. return current_step * self.step_load, self.spawn_rate 43. 44. 45. if __name__ == '__main__': 46. file_path = os.path.abspath(__file__) 47. os.system(f'locust -f {file_path} --web-host=127.0.0.1')
基于时间阶段负载策略
- 前10s和10-20s用户数为10；20-50s用户数为50；50-80s用户数为100；80s后用户数为30
1. # -*- coding: utf-8 -*- 2. # @Time : 2021/5/2 3. # @Author : 大海 4. 5. import os 6. from locust import HttpUser, TaskSet, task, constant 7. from locust import LoadTestShape 8. 9. 10. class UserTasks(TaskSet): 11. @task 12. def my_task(self): 13. self.client.get("/") 14. 15. 16. class WebsiteUser(HttpUser): 17. wait_time = constant(0.5) 18. tasks = [UserTasks] 19. host = 'https://baidu.com' 20. 21. 22. class TimeShape(LoadTestShape): 23. """ 24. duration -- 多少秒后进入下一个阶段 25. users -- 用户数 26. spawn_rate -- 每秒要启动/停止的用户数 27. """ 28. 29. stages = [ 30. {"duration": 10, "users": 10, "spawn_rate": 10}, 31. {"duration": 20, "users": 50, "spawn_rate": 10}, 32. {"duration": 50, "users": 100, "spawn_rate": 10}, 33. {"duration": 80, "users": 30, "spawn_rate": 10} 34. ] 35. 36. def tick(self): 37. run_time = self.get_run_time() 38. 39. for stage in self.stages: 40. if run_time < stage["duration"]: 41. tick_data = (stage["users"], stage["spawn_rate"]) 42. return tick_data 43. 44. return None 45. 46. 47. if __name__ == '__main__': 48. file_path = os.path.abspath(__file__) 49. os.system(f'locust -f {file_path} --web-host=127.0.0.1')
