如何修复Locust自定义消息示例中的'WorkerNodes无encode属性'错误

问题描述 #

我正在做分布式负载生成，需要协调Worker节点间的数据传递，参考了Locust的跨节点通信示例实现逻辑，但运行Master节点时出现了如下错误：

[2025-01-29 11:50:24,427] f3d0cd3a1386/INFO/locust.runners: Sending spawn jobs of 1 users at 1.00 spawn rate to 1 ready workers
[2025-01-29 11:50:24,428] f3d0cd3a1386/INFO/locust.util.exception_handler: Exception found on retry 1: -- retry after 1s
[2025-01-29 11:50:24,428] f3d0cd3a1386/ERROR/locust.util.exception_handler: 'WorkerNodes' object has no attribute 'encode'
Traceback (most recent call last):
  File "/opt/venv/lib/python3.12/site-packages/locust/util/exception_handler.py", line 13, in wrapper
    return function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.12/site-packages/locust/rpc/zmqrpc.py", line 34, in send_to_client
    self.socket.send_multipart([msg.node_id.encode(), msg.serialize()])
                                ^^^^^^^^^^^^^^^^^^
AttributeError: 'WorkerNodes' object has no attribute 'encode'
[2025-01-29 11:50:25,431] f3d0cd3a1386/INFO/locust.util.exception_handler: Exception found on retry 2: -- retry after 3s
[2025-01-29 11:50:25,432] f3d0cd3a1386/ERROR/locust.util.exception_handler: 'WorkerNodes' object has no attribute 'encode'
Traceback (most recent call last):
  File "/opt/venv/lib/python3.12/site-packages/locust/util/exception_handler.py", line 13, in wrapper
    return function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.12/site-packages/locust/rpc/zmqrpc.py", line 34, in send_to_client
    self.socket.send_multipart([msg.node_id.encode(), msg.serialize()])
                                ^^^^^^^^^^^^^^^^^^
AttributeError: 'WorkerNodes' object has no attribute 'encode'
[2025-01-29 11:50:28,434] f3d0cd3a1386/INFO/locust.util.exception_handler: Exception found on retry 3: -- retry after 5s
[2025-01-29 11:50:28,435] f3d0cd3a1386/ERROR/locust.util.exception_handler: 'WorkerNodes' object has no attribute 'encode'
Traceback (most recent call last):
  File "/opt/venv/lib/python3.12/site-packages/locust/util/exception_handler.py", line 13, in wrapper
    return function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.12/site-packages/locust/rpc/zmqrpc.py", line 34, in send_to_client
    self.socket.send_multipart([msg.node_id.encode(), msg.serialize()])
                                ^^^^^^^^^^^^^^^^^^
AttributeError: 'WorkerNodes' object has no attribute 'encode'
[2025-01-29 11:50:33,437] f3d0cd3a1386/INFO/locust.util.exception_handler: Retry failed after 3 times.
[2025-01-29 11:50:33,442] f3d0cd3a1386/ERROR/root: Uncaught exception in event handler: 
Traceback (most recent call last):
  File "/opt/venv/lib/python3.12/site-packages/locust/event.py", line 47, in fire
    handler(**kwargs)
  File "/mnt/locust/locustfile.py", line 65, in on_test_start
    environment.runner.send_message("test_users", data, worker)
  File "/opt/venv/lib/python3.12/site-packages/locust/runners.py", line 1185, in send_message
    self.server.send_to_client(Message(msg_type, data, client_id))
  File "/opt/venv/lib/python3.12/site-packages/locust/util/exception_handler.py", line 13, in wrapper
    return function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.12/site-packages/locust/rpc/zmqrpc.py", line 34, in send_to_client
    self.socket.send_multipart([msg.node_id.encode(), msg.serialize()])
                                ^^^^^^^^^^^^^^^^^^
AttributeError: 'WorkerNodes' object has no attribute 'encode'

[2025-01-29 11:50:34,444] f3d0cd3a1386/INFO/locust.runners: Spawning is complete and report waittime is expired, but not all reports received from workers: {"WebsiteUser": 0} (0 total users)

排查后发现问题出在这行代码：

environment.runner.send_message("test_users", data, worker)

移除worker参数后错误消失，但我需要给不同Worker节点发送不同的数据，请问该如何修复？

解决方案 #

错误原因 #

send_message方法的第三个参数需要传入字符串类型的Worker节点ID，但你传入的是WorkerNodes对象，代码尝试调用该对象的encode()方法（用于将字符串转为字节流）时失败，因为只有字符串才有这个方法。

修复方式 #

方式1：传入Worker对象的id属性

如果你的worker变量是从事件参数中获取的Worker实例，直接提取它的id属性作为第三个参数即可：

# 正确写法：传入worker.id而非worker对象
environment.runner.send_message("test_users", data, worker.id)

方式2：遍历所有Worker节点发送对应数据

如果需要给每个Worker分配不同的数据，可以先获取所有Worker的ID，再循环发送：

def on_test_start(environment, **kwargs):
    # 自定义：按Worker ID分配对应的数据
    worker_data_map = {
        "worker-node-1": {"task_group": "group_a", "user_count": 15},
        "worker-node-2": {"task_group": "group_b", "user_count": 25}
    }
    
    # 遍历所有已连接的Worker节点
    for worker_id in environment.runner.clients.keys():
        # 取出当前Worker对应的数据，无匹配则用默认值
        target_data = worker_data_map.get(worker_id, {"task_group": "default"})
        # 发送消息到指定Worker
        environment.runner.send_message("test_users", target_data, worker_id)

验证 #

修改后重启Master和Worker节点，错误会消失，且每个Worker能收到对应的数据。

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