Locust高并发情况下的性能优化与分布式场景的应用

在使用Locust过程时，有时我们会发现当进行高并发压测时，得到的RPS往往会比Jmeter等工具得到的结果更低。

那究竟是什么原因呢？

本文将会针对该问题进行分析并给出解决方式。

问题描述

最近在压测过程中，为了验证Locust本身压测结果准确，我们将其与传统压测工具Jmeter进行比较。

然而，在压测结果比较过程中，我们发现当Jmeter在100并发时可以达到500qps，而利用Locust进行压测时，同样是在100并发时，发现得到的结果只能达到300qps左右。
那么问题出现在哪儿呢？

原因分析

通过查阅大量的资料发现，Locust的client本身是基于python的第三方库requests，然而，requests本身为了简化requests包的使用的便捷，造成了requests的包相对的资源消耗更高。
因此，由于requests本身资源较高，导致发压工具本身的性能消耗过高，从而导致最终的数据并不准备。

解决方式

那么，具体应该如何解决呢？
我们可以通过如下两个方面来进行优化：

使用分布式Locust工具进行施压（提高机器的配置）
使用Locust的geventhttpclient版本。

首先，我们来讨论方式1：使用分布式Locust工具进行施压并提高机器配置。
显而易见，当我们机器资源提升并使用多机进行分压时，可以保证我们每个机器的资源没有打满，此时，requests包的资源消耗则不会产生较大的影响。

其次，Locust的geventhttpclient版本是专门针对大并发的情况下，requests包性能的优化。在该分支中，Locust的client包没有直接使用requests这个高资源消耗的库，相反，而是直接基于原生的urllib2库进行开发，大幅度降低了资源的消耗。

结论

事实证明，在使用Locust的geventhttpclient版本后，在2核4G的Linux机器中，单机并发在500rps时，性能和Jmeter基本一致。
在超过500rps时，则建议进行多机分布式施压，平均每个实例分压在500rps即可。
Ps：需要注意的是，在进行多机分布式施压时，Master节点仅用于控制，而没有用于发压。

# -*- coding: utf-8 -*-
# @Time    : 2021/6/11 10:42
# @Author  : ward
# @File    : locust_1.py

from locust import HttpUser, TaskSet, task, events
import time, json, os, sys, socket
from locust.exception import LocustError
from geventhttpclient import HTTPClient
from geventhttpclient.url import URL

host = "http://xxx.cn"


# 定义用户行为，继承TaskSet类，用于描述用户行为
# (这个类下面放各种请求，请求是基于requests的，每个方法请求和requests差不多，请求参数、方法、响应对象和requests一样的使用，url这里写的是路径)
# client.get===>requests.get
# client.post===>requests.post

class test_126(TaskSet):
    # wait_time = between(5, 15)
    # task装饰该方法为一个事务方法的参数用于指定该行为的执行权重。参数越大，每次被虚拟用户执行概率越高，不设置默认是1，
    # @task(1)
    # def test_api_1(self):
    #     # 定义requests的请求头
    #     header = {"User-Agent": "Mozilla/5.0 "
    #                             "(Windows NT 6.1; Win64; x64) AppleWebKit/537.36 "
    #                             "(KHTML, like Gecko) Chrome/60.0.3112.101 Safari/537.36"}
    #     # r是包含所有响应内容的一个对象
    #     r = self.client.get("/union/lily/v1/getMenu?allianceid=23&sitetype=pc", timeout=30, headers=header)
    #     # 这里可以使用assert断言请求是否正确，也可以使用if判断
    #     if r.status_code == 200:

    # print("success")

    # @task(2)
    # def test_api_2(self):
    #     # 定义requests的请求头
    #     header = {"User-Agent": "Mozilla/5.0 "
    #                             "(Windows NT 6.1; Win64; x64) AppleWebKit/537.36 "
    #                             "(KHTML, like Gecko) Chrome/60.0.3112.101 Safari/537.36"}
    #     # r是包含所有响应内容的一个对象
    #     r = self.client.get("/union/lily/v1/getInfo?allianceid=23&infoid=187013", timeout=30, headers=header)
    #     # 这里可以使用assert断言请求是否正确，也可以使用if判断
    #     if r.status_code == 200:
    #         print("success")

    def on_start(self):
        '''初始化数据'''
        url = URL(host)
        # 若为https请求，ssl设置为True
        self.http = HTTPClient(url.host, url.port, ssl=False, connection_timeout=20, network_timeout=20)

    def on_stop(self):
        '''运行结束，关闭http/https连接'''
        self.http.close()

    @task
    def test_api_2(self):
        start_time = time.time()
        try:
            # 定义requests的请求头

            header = {"User-Agent": "Mozilla/5.0 "
                                    "(Windows NT 6.1; Win64; x64) AppleWebKit/537.36 "
                                    "(KHTML, like Gecko) Chrome/60.0.3112.101 Safari/537.36"}
            # r是包含所有响应内容的一个对象
            r = self.http.get("/union/lily/v1/getInfo?allianceid=23&infoid=187013", headers=header)
            # 这里可以使用assert断言请求是否正确，也可以使用if判断
            if r.status_code == 200:
                data = r.read()
                # print(data)
                end_time = time.time()
                response_time = int((end_time - start_time) * 1000)
                response_length = len(data)
                # print(r.__dict__)
                print("success")
                events.request_success.fire(request_type="GET", name="test_success", response_time=response_time,
                                            response_length=response_length)

        except AssertionError:
            end_time = time.time()
            response_time = int((end_time - start_time) * 1000)
            events.request_failure.fire(request_type="GET", name="test_failure", response_time=response_time,
                                        response_length=0,
                                        exception="断言错误。status_code：{}。接口返回：。".format(r.status_code, ),
                                        tb=sys.exc_info()[2])

        except socket.timeout:
            end_time = time.time()
            # events.user_error()  locust_error
            response_time = int((end_time - start_time) * 1000)
            events.request_failure.fire(request_type="GET", name="test_failure", response_time=response_time,
                                        response_length=0, exception='Timeout', tb=sys.exc_info()[2])

        except Exception as e:
            end_time = time.time()
            # events.user_error()  locust_error
            response_time = int((end_time - start_time) * 1000)
            events.request_failure.fire(request_type="GET", name="test_failure", response_time=response_time,
                                        response_length=0, exception='error', tb=sys.exc_info()[2])


# 这个类类似设置性能测试，继承HttpLocust
class websitUser(HttpUser):
    host = "http://xxx.cn"
    # 指向一个上面定义的用户行为类
    tasks = [test_126]
    # 执行事物之间用户等待时间的下界，单位毫秒，相当于lr中的think time
    min_wait = 1000
    max_wait = 3000

# if __name__ == '__main__':
#     import os
#
#     os.system('locust -f locust_2.py --host=http://xxx.cn  --web-host="127.0.0.1"')