如何在Locust的多个任务类之间共享或传递变量？

我正尝试使用Locust编写一个简单的性能测试任务，原本两个顺序执行的HTTP任务在同一个类中运行正常，但我希望将任务拆分为两个SequentialTaskSet类，需要在第二个类中使用第一个类的变量。该如何实现？

# FIRST CLASS
from locust import SequentialTaskSet

class MySeqTask1(SequentialTaskSet):

    @task
    def get_status1(self):
        r = self.client.get("/200")
        print("1. Status of 200")
        self.data = 20 # ***需要共享到第二个类的变量***

# SECOND CLASS 
from locust import SequentialTaskSet, HttpUser, constant, task, FastHttpUser
from sampleLocust import MySeqTask1
    
class MySeqTask2(SequentialTaskSet):

    @task
    def get_status3(self):
        obj = MySeqTask1(self)
        self.client.get(f"/301")
        print("3. Status of 301")
        print(obj.data) # ***使用第一个类的变量***

    class MyLoadTest(FastHttpUser):
       host = "https://http.cat"
       tasks = [MySeqTask1, MySeqTask2]
       wait_time = constant(1)

运行时遇到以下错误：

运行命令：locust -f utils/Baseclassone.py --headless -u 1 -r 1 -t 1m

print(obj.data) # ***USING VARIABLE FROM FIRST CLASS***
AttributeError: 'MySeqTask1' object has no attribute 'data'

问题原因 #

你手动实例化MySeqTask1的方式完全错误：Locust的TaskSet实例由User类统一管理，手动创建的实例并没有执行get_status1任务，自然不会生成data属性。另外tasks = [MySeqTask1, MySeqTask2]的写法会让User随机执行两个TaskSet，无法保证你需要的顺序执行逻辑。

正确实现方式 #

跨TaskSet共享变量的核心是把数据存储到所属的User实例上（TaskSet通过self.user属性关联到User），同时如果需要顺序执行两个TaskSet，必须用外层SequentialTaskSet包裹来保证执行顺序。

方案1：顺序执行TaskSet并通过User共享变量

from locust import SequentialTaskSet, FastHttpUser, constant, task

class MySeqTask1(SequentialTaskSet):
    @task
    def get_status1(self):
        r = self.client.get("/200")
        print("1. Status of 200")
        # 将变量存储到User实例，实现跨TaskSet共享
        self.user.shared_data = 20
        # 执行完当前TaskSet后切换到下一个
        self.interrupt()

class MySeqTask2(SequentialTaskSet):
    @task
    def get_status3(self):
        self.client.get(f"/301")
        print("3. Status of 301")
        # 从User实例读取共享变量
        print(self.user.shared_data)
        self.interrupt()

class MyLoadTest(FastHttpUser):
    host = "https://http.cat"
    wait_time = constant(1)
    # 用外层SequentialTaskSet包裹，保证两个TaskSet顺序执行
    class MainTaskSet(SequentialTaskSet):
        tasks = [MySeqTask1, MySeqTask2]
    tasks = [MainTaskSet]

方案2：无严格顺序时的变量共享

如果不需要严格顺序执行，仅需共享变量，同样把数据存在User实例上，同时可以给变量设置默认值避免异常：

from locust import SequentialTaskSet, FastHttpUser, constant, task

class MySeqTask1(SequentialTaskSet):
    @task
    def get_status1(self):
        r = self.client.get("/200")
        print("1. Status of 200")
        self.user.shared_data = 20

class MySeqTask2(SequentialTaskSet):
    @task
    def get_status3(self):
        # 先检查User是否存在该属性，避免未执行Task1就执行Task2的报错
        if hasattr(self.user, 'shared_data'):
            print(self.user.shared_data)
        self.client.get(f"/301")
        print("3. Status of 301")

class MyLoadTest(FastHttpUser):
    host = "https://http.cat"
    tasks = [MySeqTask1, MySeqTask2]
    wait_time = constant(1)
    # 在User初始化时给共享变量设默认值
    def __init__(self, environment):
        super().__init__(environment)
        self.shared_data = None

关键要点 #

• 所有跨TaskSet的共享数据必须存在self.user上，这是不同TaskSet的唯一公共关联对象。
• 手动实例化TaskSet是错误操作，TaskSet的生命周期由User类管理，无需手动创建。
• 要保证顺序执行多个TaskSet，必须用外层SequentialTaskSet包裹，否则Locust会随机选择tasks列表中的TaskSet执行。

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