为什么要进行负载测试

Locust 是什么

• 使用 Python 编写测试脚本

安装

pip install locust

easy_install locust

locust --help

快速入门

from locust import HttpUser, TaskSet

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

def index(l):    l.client.get("/")

def profile(l):    l.client.get("/profile")

class UserBehavior(TaskSet):    tasks = {index: 2, profile: 1}
    def on_start(self):        login(self)

class WebsiteUser(HttpUser):    tasks = [UserBehavior]    min_wait = 5000    max_wait = 9000

from locust import HttpUser, 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(HttpUser):    tasks = [UserBehavior]    min_wait = 5000    max_wait = 9000

locust --host=http://example.com

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

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

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

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

编写 locustfile

from locust import User, TaskSet, task

class MyTaskSet(TaskSet):    @task    def my_task(self):        print("executing my_task")

class MyUser(User):    tasks = [MyTaskSet]    min_wait = 5000    max_wait = 15000

locust -f locust_file.py WebUser MobileUser

class WebUser(User):    weight = 3    ....

class MobileUser(User):    weight = 1    ....

from locust import User, TaskSet, task

class MyTaskSet(TaskSet):    min_wait = 5000    max_wait = 15000
    @task(3)    def task1(self):        pass
    @task(6)    def task2(self):        pass

class MyUser(User):    tasks = [MyTaskSet]

from locust import User, TaskSet

def my_task(l):    pass

class MyTaskSet(TaskSet):    tasks = [my_task]

class MyUser(User):    tasks = [MyTaskSet]

{my_task: 3, another_task: 1}

Main user behaviour  Index page
  Forum page  Read thread    Reply  New thread  View next page
  Browse categories    Watch movie    Filter movies  About page

class ForumPage(TaskSet):    @task(20)    def read_thread(self):        pass
    @task(1)    def new_thread(self):        pass
    @task(5)    def stop(self):        self.interrupt()

class UserBehaviour(TaskSet):    tasks = {ForumPage: 10}
    @task    def index(self):        pass

class MyTaskSet(TaskSet):    @task    class SubTaskSet(TaskSet):        @task        def my_task(self):            pass

class HttpUser:    代表被“孵化”出来的、用于“攻击”要进行负载测试的系统的 HTTP 用户。    用户行为由 tasks 属性来定义。    这个类在初始化时，将创建 client 属性，client 属性是支持在请求之间保持用户会话的 HTTP 客户端。
    client = None

from locust import HttpUser, TaskSet, task

class MyTaskSet(TaskSet):    @task(2)    def index(self):        self.client.get("/")
    @task(1)    def about(self):        self.client.get("/about/")

class MyUser(HttpUser):    tasks = [MyTaskSet]    min_wait = 5000    max_wait = 15000

response = self.client.post("/login", {"username": "testuser", "password": "secret"})

with client.get("/", catch_response=True) as response:    if response.content != "Success":        response.failure("Got wrong response")

with client.get("/does_not_exist/", catch_response=True) as response:    if response.status_code == 404:        response.success()

# Statistics for these requests will be grouped under: /blog/?id=[id]for i in range(10):    client.get("/blog?id=%i" % i, name="/blog?id=[id]")

分布式地运行 Locust

locust -f my_locustfile.py --master

locust -f my_locustfile.py --worker --master-host=192.168.0.14

测试非 HTTP 系统

import randomimport timefrom xmlrpc.server import SimpleXMLRPCServer

def get_time():    time.sleep(random.random())    return time.time()

def get_random_number(low, high):    time.sleep(random.random())    return random.randint(low, high)

server = SimpleXMLRPCServer(("localhost", 8877))print("Listening on port 8877...")server.register_function(get_time, "get_time")server.register_function(get_random_number, "get_random_number")server.serve_forever()

import timefrom xmlrpc.client import ServerProxy, Fault
from locust import User, task

class XmlRpcClient(ServerProxy):    """    XmlRpcClient is a wrapper around the standard library's ServerProxy.    It proxies any function calls and fires the *request* event when they finish,    so that the calls get recorded in Locust.    """
    def __init__(self, host, request_event):        super().__init__(host)        self._request_event = request_event
    def __getattr__(self, name):        func = ServerProxy.__getattr__(self, name)
        def wrapper(*args, **kwargs):            request_meta = {                "request_type": "xmlrpc",                "name": name,                "start_time": time.time(),                "response_length": 0,  # calculating this for an xmlrpc.client response would be too hard                "response": None,                "context": {},  # see HttpUser if you actually want to implement contexts                "exception": None,            }            start_perf_counter = time.perf_counter()            try:                request_meta["response"] = func(*args, **kwargs)            except Fault as e:                request_meta["exception"] = e            request_meta["response_time"] = (time.perf_counter() - start_perf_counter) * 1000            self._request_event.fire(**request_meta)  # This is what makes the request actually get logged in Locust            return request_meta["response"]
        return wrapper

class XmlRpcUser(User):    """    A minimal Locust user class that provides an XmlRpcClient to its subclasses    """
    abstract = True  # dont instantiate this as an actual user when running Locust
    def __init__(self, environment):        super().__init__(environment)        self.client = XmlRpcClient(self.host, request_event=environment.events.request)

# The real user class that will be instantiated and run by Locust# This is the only thing that is actually specific to the service that we are testing.class MyUser(XmlRpcUser):    host = "http://127.0.0.1:8877/"
    @task    def get_time(self):        self.client.get_time()
    @task    def get_random_number(self):        self.client.get_random_number(0, 100)

事件钩子

from locust import events
@events.request.add_listenerdef my_request_handler(request_type, name, response_time, response_length, response,                       context, exception, start_time, url, **kwargs):    if exception:        print(f"Request to {name} failed with exception {exception}")    else:        print(f"Successfully made a request to: {name}")        print(f"The response was {response.text}")

增加 Web 路由

from locust import events
@events.init.add_listenerdef on_locust_init(environment, **kw):    @environment.web_ui.app.route("/added_page")    def my_added_page():        return "Another page"