Locust là một framework kiểm thử hiệu năng mã nguồn mở, được thiết kế để mô phỏng hàng nghìn đến hàng triệu người dùng ảo tương tác đồng thời với hệ thống web hoặc API. Khác biệt lớn so với các công cụ truyền thống như JMeter, Locust sử dụng tiếp cận dựa trên hành vi người dùng thực tế — mỗi "người dùng" được biểu diễn dưới dạng một coroutine Python chạy trong môi trường gevent, cho phép đạt mật độ kết nối cao trên một máy duy nhất mà không cần cấu hình phức tạp.

Tại sao chọn Locust?

• Mã hóa bằng Python thuần: Không cần học ngôn ngữ đặc thù hay XML — chỉ cần viết hàm và gắn decorator.
• Hỗ trợ phân tán natively: Kết hợp master-worker qua ZeroMQ để mở rộng quy mô kiểm thử vượt xa giới hạn phần cứng đơn lẻ.
• Giao diện giám sát trực quan: Dashboard web thời gian thực hiển thị RPS, latency, tỉ lệ thất bại và phân bố phản hồi.
• Tính linh hoạt cao: Không giới hạn ở HTTP — có thể tích hợp với gRPC, WebSocket, cơ sở dữ liệu hoặc bất kỳ giao thức nào thông qua thư viện Python tương ứng.

Cài đặt và chuẩn bị

Cài đặt chính thức từ PyPI:

pip install locust

Nếu triển khai theo mô hình master-worker, cài thêm pyzmq:

pip install pyzmq

Cấu trúc cơ bản của kịch bản kiểm thử

Một tệp locustfile.py gồm ba thành phần cốt lõi:

1. TaskSet: Lớp định nghĩa chuỗi hành động (ví dụ: đăng nhập → xem danh sách → click vào chi tiết).
2. HttpUser (thay thế cho HttpLocust từ phiên bản 2.0+): Lớp đại diện cho một người dùng ảo, chứa client để gửi yêu cầu HTTP.
3. Các hàm tác vụ: Có thể khai báo qua decorator @task hoặc gán vào thuộc tính tasks dưới dạng từ điển trọng số.

Ví dụ minh họa: Kịch bản người dùng đa vai trò

from locust import HttpUser, TaskSet, task, between
import random

class ProductBrowsing(TaskSet):
    @task(3)
    def view_catalog(self):
        self.client.get("/api/products", name="Danh mục sản phẩm")

    @task(1)
    def search_by_keyword(self):
        keyword = random.choice(["laptop", "mouse", "keyboard"])
        self.client.get(f"/api/search?q={keyword}", name="Tìm kiếm sản phẩm")

class AuthFlow(TaskSet):
    def on_start(self):
        self.login()

    def login(self):
        self.client.post(
            "/auth/login",
            json={"email": "test@example.com", "password": "pass123"},
            name="Đăng nhập"
        )

    @task(2)
    def access_profile(self):
        self.client.get("/api/me", name="Truy cập hồ sơ")

    @task
    def logout(self):
        self.client.post("/auth/logout", name="Đăng xuất")

class WebTester(HttpUser):
    host = "https://demo-api.example.com"
    wait_time = between(2, 5)
    tasks = [ProductBrowsing, AuthFlow]

Trong ví dụ trên:

• Mỗi HttpUser khởi tạo một phiên kết nối HTTP bền (giữ session cookie, header chung…).
• wait_time = between(2, 5) đảm bảo mỗi người dùng chờ ngẫu nhiên từ 2–5 giây giữa các tác vụ.
• Hai lớp TaskSet được gán vào tasks với ưu tiên ngẫu nhiên — ProductBrowsing chiếm ~75% tổng hành vi do trọng số cao hơn.
• Phương thức on_start() chỉ chạy một lần khi người dùng bắt đầu, phù hợp cho thao tác khởi tạo như đăng nhập.

Chạy kiểm thử

Từ thư mục chứa locustfile.py, thực thi:

locust --host=https://demo-api.example.com

Sau đó truy cập http://localhost:8089 để mở dashboard. Bạn có thể cấu hình số người dùng ảo, tốc độ tăng dần (hatch rate) và chọn chế độ phân tán nếu cần.

Xử lý phản hồi nâng cao

Để kiểm soát lỗi tùy chỉnh (ví dụ: chấp nhận mã 404 như một kết quả hợp lệ), dùng tham số catch_response=True:

with self.client.get("/api/v1/legacy-endpoint", catch_response=True) as resp:
    if resp.status_code == 404:
        resp.success()  # Đánh dấu là thành công dù HTTP status là 404
    elif resp.status_code != 200:
        resp.failure(f"Unexpected status: {resp.status_code}")

Định danh yêu cầu động

Khi URL chứa tham số biến đổi (ví dụ ID sản phẩm), dùng tham số name để nhóm thống kê:

product_id = random.randint(1001, 9999)
self.client.get(f"/api/products/{product_id}", name="/api/products/[id]")

Điều này giúp Locust gộp tất cả các request tới /api/products/1234, /api/products/5678… vào cùng một dòng trong báo cáo.