为什么这个泡菜不吃？

Question

我试图将它作为多进程进程来运行，但是当启动线程时，pickler会掉下来，我无法弄清楚是什么阻止了它的酸洗。我尝试过注释套接字代码和消息obj代码，但仍然不能工作--我做错了什么？

class TransmitThread(Process):

    def __init__(self, send_queue, reply_queue, control_pipe, recv_timeout=2, buffer_size=4096):
        """
            This init function is called when the thread is created. Function simply calls the Process class init
            function, and stores the class vars.
        """
        # Call process class init
        Process.__init__(self)

        # Store class vars
        self.send_queue     = send_queue
        self.reply_queue    = reply_queue
        self.control_pipe   = control_pipe
        self._recv_timeout  = recv_timeout
        self._buffer_size   = buffer_size

    def run(self):
        """
            This is the main function that is called when the thread is started.
            The function loops forever, waiting for a send message in the queue, and processes the message to send
            and fetches the response. The thread loops forever until it's terminated or the KILL THREAD command is
            passed through the control pipe.
        """
        # Start our forever running loop
        while True:

            # Check if there is anything in the pipe
            if self.control_pipe.poll():
                # Check if we received the kill thread command
                if self.control_pipe.recv() == KILL_THREAD_COMMAND:
                    # Kill the while loop and end the thread
                    break

            # Check if there is anything in message queue
            if not self.send_queue.empty():
                # Fetch message from the queue to send, and unpickle
                message_obj, message_pickle = self.send_queue.get()

                # Open socket and set timeout
                sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                sock.settimeout(self._recv_timeout)

                # Connect socket to the recipient
                sock.connect( message_obj.recipient_address )

                # Push the pickled message down the socket
                sock.sendall(message_pickle)

                # Check if the message we send is a request (should get a response)
                if str(message_obj.message_type) == str(Message.REQUEST):

                    print "fetching reply"

                    # Lets fetch the response, and push the pickled message onto the queue
                    self.reply_queue.put( sock.recv(self._buffer_size) )

                    print "got a reply"

                # All done, close the socket
                sock.close()

            # Add small delay to stop this thread consuming too much CPU time
            sleep(0.1)

错误信息是：

File "C:/Users/oliver/OneDrive/GIT/pyke/pyke.py", line 127, in __init__
    self.thread.start()
  File "C:\Python27\lib\multiprocessing\process.py", line 130, in start
    self._popen = Popen(self)
  File "C:\Python27\lib\multiprocessing\forking.py", line 277, in __init__
    dump(process_obj, to_child, HIGHEST_PROTOCOL)
  File "C:\Python27\lib\multiprocessing\forking.py", line 199, in dump
    ForkingPickler(file, protocol).dump(obj)
  File "C:\Python27\lib\pickle.py", line 224, in dump
    self.save(obj)
  File "C:\Python27\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Python27\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Python27\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 725, in save_inst
    save(stuff)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Python27\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Python27\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Python27\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Python27\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Python27\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Python27\lib\pickle.py", line 396, in save_reduce
    save(cls)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 748, in save_global
    (obj, module, name))
pickle.PicklingError: Can't pickle <type 'thread.lock'>: it's not found as thread.lock
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Python27\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Python27\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Python27\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Python27\lib\pickle.py", line 880, in load_eof
    raise EOFError
EOFError

Answer 1

pickler错误来自于multiprocessing.Process试图在内部对子进程进行筛选。我非常肯定，您的一个实例变量没有正确地选择子进程。从你的问题看哪一个不清楚？

    # Store class vars
    self.send_queue     = send_queue
    self.reply_queue    = reply_queue
    self.control_pipe   = control_pipe
    self._recv_timeout  = recv_timeout
    self._buffer_size   = buffer_size

在OP的注释后编辑

问题是send_queue和reply_queue是Queue.Queue而不是multiprocessing.Queue。在分叉子工作人员时，Process尝试将自身和任何实例变量序列化到子实例。但是，Queue.Queue是不可序列化的本地对象，因此出现了错误。

与这个问题相关的还有这样一个事实，即multiprocessing.Queue 重新使用异常来自Queue，没有再导出这些信息。这个被记录在案，尽管有点隐藏在杂乱的下面：

注意： multiprocessing使用通常的Queue.Empty和Queue.Full异常来表示超时。它们在multiprocessing名称空间中不可用，因此您需要从Queue导入它们。