<class '_pickle.UnpicklingError'> - 无效的加载键，ZMQ socket.recv_pyobj() 中的 '\x00'

Question

我已参考相关查询，无法修复上述错误。我正在两个单独的进程之间运行服务器和客户端 python 对象共享。客户端需要将字典或元组发送到服务器。 我尝试将协议更改为 pickle.DEFAULT_PROTOCOL 和 pickle.HIGHEST_PROTOCOL。问题仍然没有解决...但是 out 可以打印我发送的字典。

服务器代码：

import zmq
import socket
import sys
import datetime
import pickle

def errorhandling(name: str, msg: str) -> None:
        """
        To handle Error in the API and Strategies, if you call this
        """
        try:
            print(f" {datetime.datetime.now().time()}| ERROR |{name} | {msg} | {sys.exc_info()[0]} -"
                  f" {sys.exc_info()[1]} | Line No:{sys.exc_info()[2].tb_lineno}")
                          
        except Exception as e:
            print("error", f" {name} | ErrorHandlingError - {e}")

ip = socket.gethostbyname(socket.gethostname())

ordcontext = zmq.Context()
ordsocket = ordcontext.socket(zmq.XREP)
ordsocket.bind(f"tcp://{ip}:7712")

def ordersocket():
    print("Order socket Running")
    while True:
        try:
            #orddata = self.ordsocket.recv().decode('utf-8', 'ignore')
            orddata = ordsocket.recv_pyobj()
            #orddata2 = pickle.loads(orddata)
            print(orddata)
        except Exception:
            errorhandling(name="ordername",
                          msg="ordersocket")
                
    print("Order socket NOT Running")
            
ordersocket()

客户代码：

import zmq
import time
import socket
import pickle
ip = socket.gethostbyname(socket.gethostname())
orcontext = zmq.Context()
orsocket = orcontext.socket(zmq.XREQ)
orsocket.connect(f"tcp://{ip}:7712")

def sendorder():
    while True:
        data = {"A": "Nameofstg", "LTP": 1200, "QTY":200, "SS":"Now", "PP":230}
        orsocket.send_pyobj(obj=data,protocol=pickle.HIGHEST_PROTOCOL)
        time.sleep(3)
        
sendorder()

错误（带有打印语句）：

<class '_pickle.UnpicklingError'> - invalid load key, '\x00'. 

11:34:49.665037|错误 |订单名称 |订单套接字 | - 无效的加载键，'\x00'。 |线号：35 {'A'：'Nameofstg'，'LTP'：1200，'QTY'：200，'SS'：'现在'，'PP'：230}

Answer 1

我不确定你的代码有什么问题，但试试下面的代码，它可以工作。你其实不需要pickle，socket.send_pyobj可以用来像字典一样发送对象。

假设master.py和worker.py在同一个dir下。

master.py

import zmq
import sys
import pexpect 
from getpass import getuser
from socket import gethostname


python = sys.executable
context = zmq.Context()
socket = context.socket(zmq.REP)
port = socket.bind_to_random_port("tcp://*")
server_host = gethostname()
server_socket = "%s:%s" % (server_host, port)

worker_command = "%s -m worker" % python + " %i " + server_socket
children = []
for ind in range(4):
    child = pexpect.spawn(worker_command % ind)
    children.append(child)
           
active = 4
while True:
    message = socket.recv_pyobj()
    print(message)
    socket.send_pyobj('task done')
    active -= 1
    if active == 0:
        break

和worker.py

import os
import sys
import zmq
from socket import gethostname

    
if __name__ == '__main__':
      
      
    import logging

    logging.basicConfig(filename = f"log")
    try:
        server_socket = sys.argv[-1]
        worker_id = int(sys.argv[-2])
        port_number = int(server_socket.split(':')[-1])
        
        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect('tcp://%s' % server_socket)
        
        data = {"A": "Nameofstg_%i" % worker_id}
        socket.send_pyobj(data)
        socket.recv_pyobj()
    except Exception as e:
        logging.exception(e)

运行python master.py可以看到，四个worker发送的字典都可以被master成功接收到，实际上你不需要pickle，因为我们使用socket.send_pyobj。

您可能还注意到，在worker.py 中，我使用logging.basicConfig(filename = f"log")。这样做可以捕获工作进程中的Exception。