Python 경량 성능 도구 Locust 설치 및 사용 방법

Locust는 스레드 프로세스의 한계를 깨고 테스트 머신에서 높은 동시성을 실행할 수 있는 Python의 코루틴 메커니즘을 기반으로 합니다.

성능 테스트의 기본

　1. 속도: 시스템의 처리 효율성 측정: 응답 시간

　2 . 얼마: 시스템의 처리 용량 측정: 단위 시간당 처리할 수 있는 트랜잭션 수(tps)

성능 테스트는 테스트 요구 사항에 따라 가장 일반적으로 다음 세 가지 범주로 나뉩니다.

　1 부하 테스트

지속적으로 서버에 압력을 가하거나, 예약할 만한 지표 또는 일부 시스템 리소스가 병목 현상에 도달했습니다. 시스템의 최대 부하 용량을 찾는 것이 목적입니다.

　2 스트레스 테스트

　오랜 시간 동안 지속되는 높은 부하를 통해 시스템이 안정적인지 확인합니다.

　3 동시성 테스트:

　　동시에 요청을 서버에 제출합니다. 시스템에 트랜잭션 충돌이나 잠금 업그레이드가 있는지 확인하는 것이 목적입니다.

성능 로드 모델

locust 설치

설치에 문제가 있습니다. Douban 소스를 통해 다운로드할 수 있습니다

pip install locust

locust template

기본적으로 대부분 이 템플릿을 기반으로 장면을 수정할 수 있습니다. read.py

from locust import HttpUser, TaskSet, task, tag, events
# 启动locust时运行
@events.test_start.add_listener
def setup(environment, **kwargs):
    # print("task setup")
# 停止locust时运行
@events.test_stop.add_listener
def teardown(environment, **kwargs):
    print("task teardown")
class UserBehavor(TaskSet):
    #虚拟用户启用task运行
    def on_start(self):
        print("start")
        locusts_spawned.wait()
    #虚拟用户结束task运行
    def on_stop(self):
        print("stop")
    @tag('test1')
    @task(2)
    def index(self):
        self.client.get('/yetangjian/p/17320268.html')
    @task(1)
    def info(self):
        self.client.get("/yetangjian/p/17253215.html")
class WebsiteUser(HttpUser):
    def setup(self):
        print("locust setup")
    def teardown(self):
        print("locust teardown")
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    min_wait = 3000
    max_wait = 5000

참고: 여기에서는 웹 호스트를 제공했습니다.

랑데뷰 포인트 lr_rendezvous

물론 랑데부 포인트 연산을 넣을 수도 있습니다. 위 템플릿의 설정을 입력하여 실행하세요

locusts_spawned = Semaphore()
locusts_spawned.acquire()
def on_hatch_complete(**kwargs):
    """
    select_task类的钩子函数
    :param kwargs:
    :return:
    """
    locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
n = 0
class UserBehavor(TaskSet):
    def login(self):
        global n
        n += 1
        print(f"第{n}个用户登陆")
    def on_start(self):
        self.login()
        locusts_spawned.wait()
    @task
    def test1(self):
        #catch_response获取返回
        with self.client.get("/yetangjian/p/17253215.html",catch_response=True):
            print("查询结束")
class WebsiteUser(HttpUser):
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    wait_time = between(1,3)
if __name__ == '__main__':
    os.system('locust -f read.py --web-host="127.0.0.1"')

공용 사용법

위의 두 예에서는 데코레이터 events.test_start.add_listener와 같은 일부는 부하 테스트 전후에 사용됩니다. 작업 실행 전후에 on_start 및 on_stop과 같은 일부 작업을 수행하는 데 사용할 수 있습니다. 작업 가중치 할당

대기 시간

# wait between 3.0 and 10.5 seconds after each task
#wait_time = between(3.0, 10.5)
#固定时间等待
# wait_time = constant(3)
#确保每秒运行多少次
constant_throughput（task_runs_per_second）
#确保每多少秒运行一次
constant_pacing(wait_time)

사용자 클래스에서 wait_time을 다시 작성하여 자동 정의

태그 표시

@tag('test1')
@task(2)
def index(self):
    self.client.get('/yetangjian/p/17320268.html')

작업을 표시하면 런타임에 특정 작업을 실행할 수 있습니다.

#只执行标记test1
os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"')
#不执行标记过的
os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"')
#除去test1执行所有
os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')

사용자 정의 실패

#定义响应时间超过0.1就为失败
with self.client.get("/yetangjian/p/17253215.html", catch_response=True) as response:
    if response.elapsed.total_seconds() > 0.1:
        response.failure("Request took too long")
#定义响应码是200就为失败
with self.client.get("/yetangjian/p/17320268.html", catch_response=True) as response:
    if response.status_code == 200:
        response.failure("响应码200，但我定义为失败")

사용자 정의 로드 모양

LoadTestShape를 상속하고 틱을 재정의하여 Shape.py를 자체 정의합니다.

이 모양 클래스는 100블록 단위로 사용자 수를 증가시킵니다. of 20, 그리고 10분 후에 로드 테스트를 중지합니다. (실행의 51초부터 user_count는 100으로 반올림됩니다.)

from locust import LoadTestShape
class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20
    def tick(self):
        run_time = self.get_run_time()
        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, self.spawn_rate)
        return None

실행 다이어그램은 다음과 같습니다

다르게

os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')

를 트리거하는 예 명령줄을 통한 시간 단계

from locust import LoadTestShape
class StagesShapeWithCustomUsers(LoadTestShape):
    stages = [
        {"duration": 10, "users": 10, "spawn_rate": 10},
        {"duration": 30, "users": 50, "spawn_rate": 10},
        {"duration": 60, "users": 100, "spawn_rate": 10},
        {"duration": 120, "users": 100, "spawn_rate": 10}]
    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data
        return None

위 내용은 Python 경량 성능 도구 Locust 설치 및 사용 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!