Locust自定义请求形状技术咨询：如何实现毫秒级指数分布的请求发送间隔控制

Hey, great question—this is a common need when you need fine-grained control over request pacing, and Locust absolutely supports it, even though the default examples lean on whole seconds. Here's how to pull it off:

Core Solution: Custom wait_time Function #

Locust's wait_time parameter accepts a function that returns a float representing seconds—so to get millisecond precision, you just return a decimal value (e.g., 0.012 for 12ms, 0.059 for 59ms). For exponential distribution, you can either use a library like NumPy for simplicity, or implement it with pure Python if you want to avoid extra dependencies.

Option 1: Using NumPy (Simpler) #

NumPy has a built-in exponential function that generates values following an exponential distribution. Just convert your desired mean interval from milliseconds to seconds, and return the result:

from locust import HttpUser, task
import numpy as np

class MyUser(HttpUser):
    # Set mean interval to 50ms (converted to 0.05 seconds)
    mean_interval_sec = 50 / 1000

    def wait_time(self):
        # Return exponential distribution value in seconds
        return np.random.exponential(self.mean_interval_sec)

    @task
    def my_request(self):
        self.client.get("/your-target-endpoint")

Option 2: Pure Python (No External Dependencies) #

If you don't want to use NumPy, you can generate exponential distribution values using Python's built-in random and math modules. The formula for converting a uniform random value to an exponential distribution is:

-mean * ln(1 - random_value)

Here's the implementation:

from locust import HttpUser, task
import random
import math

class MyUser(HttpUser):
    # Set mean interval to 50ms (0.05 seconds)
    mean_interval_sec = 0.05

    def wait_time(self):
        # Generate uniform random value between 0 and 1
        rand_val = random.random()
        # Calculate exponential interval in seconds
        return -self.mean_interval_sec * math.log(1 - rand_val)

    @task
    def my_request(self):
        self.client.get("/your-target-endpoint")

Key Notes #

• Precision: Python's floating-point math is more than sufficient for millisecond-level accuracy—you won't lose precision here.
• Validation: To confirm your intervals are working as expected, add a timestamp log in your task to check the time between requests:

  import time
  
  last_request_time = None
  
  @task
  def my_request(self):
      global last_request_time
      current_time = time.time()
      if last_request_time:
          interval_ms = (current_time - last_request_time) * 1000
          print(f"Request interval: {interval_ms:.2f}ms")
      last_request_time = current_time
      self.client.get("/your-target-endpoint")
  

• Performance: Both implementations are lightweight enough for high-concurrency tests. The pure Python version has slightly less overhead if you're running thousands of concurrent users.

内容的提问来源于stack exchange，提问作者Alireza