如何从Master节点向Worker节点传递Shopping对象？

问题根源是Locust分布式运行时，Master与Worker之间传递的数据必须支持序列化，而你自定义的Shopping类实例无法被默认序列化机制处理，因此报错TypeError: can not serialize 'Shopping' object。

解决方案：传递可序列化配置，在Worker端重建实例 #

不要直接传递Shopping对象，而是提取它依赖的可序列化参数（如URL、请求头、乘客/航班配置等），在Worker节点上重新初始化Shopping实例。

1. 修改Master端的on_test_start逻辑，存储配置而非实例

替换原来存储shopping实例的代码，改为存储构建Shopping所需的所有可序列化参数：

@staticmethod
@events.test_start.add_listener
def on_test_start(environment: Environment, **kwargs):
    os.environ["ENABLE_SLEEP"] = "False"
    credentials = {
        "login": environment.parsed_options.login,
        "password": environment.parsed_options.password,
        "structure_unit_code": environment.parsed_options.suid,
    }
    login = Login(base_url=environment.host)
    response = login.request_login(credentials)
    parse_xml = parsing_xml_response(response.text)
    headers = {
        "Content-Type": "application/xml",
        "Authorization": f'Bearer {parse_xml.xpath("//Token")[0].text}'}
    
    # 提取Shopping所需的配置参数（包括set_paxs、set_flight设置的相关数据）
    shopping_config = {
        "base_url": environment.parsed_options.host,
        "headers": headers,
        # 根据set_paxs、set_flight的逻辑，提取对应的可序列化数据，比如：
        "pax_data": ...,  # 比如乘客数量、类型等原始数据
        "flight_data": ... # 比如航班号、日期等原始数据
    }
    # 将配置存入环境变量，替代原来的shopping实例
    environment.parsed_options.__dict__.update({"shopping_config": shopping_config})

2. 在Worker端的用户初始化时重建Shopping实例

在WebUser的on_start方法中，读取配置并重新创建Shopping实例，同时完成乘客和航班的设置：

class WebUser(HttpUser):
    tasks = [Tasks]
    min_wait = 0
    max_wait = 0

    def on_start(self):
        # 从环境中读取配置
        config = self.environment.parsed_options.shopping_config
        # 重建Shopping实例
        self.shopping = Shopping(
            base_url=config["base_url"],
            headers=config["headers"]
        )
        # 重新执行乘客和航班设置逻辑
        set_paxs(self.environment, self.shopping)
        set_flight(self.environment, self.shopping)

    @staticmethod
    @events.test_start.add_listener
    def on_test_start(environment: Environment, **kwargs):
        os.environ["ENABLE_SLEEP"] = "False"

3. 修改任务逻辑，使用Worker端的Shopping实例

更新Tasks中的代码，引用WebUser实例上的shopping对象：

class Tasks(SequentialTaskSet):
    @task
    def shopping(self):
        with self.parent.shopping.request_airshopping(
                client=self.client, catch_response=True) as response:
            self.run_req(response, self.parent.environment.parsed_options.rsa)

关键说明 #

• 所有跨节点传递的数据必须是JSON可序列化的类型（字典、列表、字符串、数字等），自定义类实例无法直接传输。
• 这种方式的核心是在Master端完成登录等全局操作，传递结果数据；在Worker端用数据重建业务对象，既保证分布式运行的兼容性，又不影响原有业务逻辑。

内容的提问来源于stack exchange，提问作者Данил Павленко