我想动态创建多个Processes，其中每个实例都有一个队列，用于接收来自其他实例的传入消息，并且每个实例还可以创建新实例。所以我们最终得到了一个进程网络，所有进程都相互发送。每个实例都可以发送给其他实例。

下面的代码会做我想做的事:它使用 Manager.dict() 来存储队列，确保传播更新，以及 Lock() 以保护对队列的写访问。但是，当添加新队列时，它会抛出 "RuntimeError: Queue objects should only be shared between processes through inheritance"。

问题是在启动时，我们不知道最终需要多少个队列，所以我们必须动态创建它们。但是由于我们不能在构建时共享队列，所以我不知道该怎么做。

我知道一种可能性是使 queues 成为全局变量，而不是将托管变量传递给 __init__:据我所知，问题是，是对 queues 变量的添加不会传播到其他进程。

编辑 我正在研究进化算法。 EA 是一种机器学习技术。 EA 模拟一个“种群”，它通过适者生存、交叉和变异而进化。在并行 EA 中，就像这里一样，我们也有种群之间的迁移，对应于进程间通信。岛屿也可以产生新的岛屿，因此我们需要一种在动态创建的进程之间发送消息的方法。

import random, time
from multiprocessing import Process, Queue, Lock, Manager, current_process
try:
    from queue import Empty as EmptyQueueException
except ImportError:
    from Queue import Empty as EmptyQueueException

class MyProcess(Process):
    def __init__(self, queues, lock):
        super(MyProcess, self).__init__(target=lambda x: self.run(x),
                                     args=tuple())
        self.queues = queues
        self.lock = lock
        # acquire lock and add a new queue for this process
        with self.lock:
            self.id = len(list(self.queues.keys()))
            self.queues[self.id] = Queue()

    def run(self):
        while len(list(self.queues.keys())) < 10:

            # make a new process
            new = MyProcess(self.lock)
            new.start()

            # send a message to a random process
            dest_key = random.choice(list(self.queues.keys()))
            dest = self.queues[dest_key]
            dest.put("hello to %s from %s" % (dest_key, self.id))

            # receive messages
            message = True
            while message:
                try:
                    message = self.queues[self.id].get(False) # don't block
                    print("%s received: %s" % (self.id, message))
                except EmptyQueueException:
                    break

            # what queues does this process know about?
            print("%d: I know of %s" %
                  (self.id, " ".join([str(id) for id in self.queues.keys()])))

            time.sleep(1)

if __name__ == "__main__":
    # Construct MyProcess with a Manager.dict for storing the queues
    # and a lock to protect write access. Start.
    MyProcess(Manager().dict(), Lock()).start()

最佳答案

我不完全确定您的实际用例是什么。也许如果您更详细地说明为什么要让每个进程动态地生成一个具有连接队列的子进程，那么在这种情况下正确的解决方案是什么会更加清楚。

无论如何，对于这个问题，现在似乎没有真正好的方法来使用多处理动态创建管道或队列。

我认为，如果您愿意在每个进程中生成线程，您可以使用 multiprocessing.connection.Listener/Client 来回通信。我没有生成线程，而是采用了一种使用网络套接字并选择在线程之间进行通信的方法。

动态进程生成和网络套接字可能仍然不稳定，具体取决于 multiprocessing 在生成/派生新进程时如何清理文件描述符，并且您的解决方案很可能在 *nix 衍生产品上更容易工作。如果您担心套接字开销，您可以使用 unix 域套接字，以增加在多台工作机器上运行节点的复杂性为代价，使其更轻量。

无论如何，这是一个使用网络套接字和全局进程列表来完成此操作的示例，因为我无法找到让 multiprocessing 执行此操作的好方法。

import collections
import multiprocessing
import random
import select
import socket
import time


class MessagePassingProcess(multiprocessing.Process):
    def __init__(self, id_, processes):
        self.id = id_
        self.processes = processes
        self.queue = collections.deque()
        super(MessagePassingProcess, self).__init__()

    def run(self):
        print "Running"
        inputs = []
        outputs = []
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        address = self.processes[self.id]["address"]
        print "Process %s binding to %s"%(self.id, address)
        server.bind(address)
        server.listen(5)
        inputs.append(server)
        process = self.processes[self.id]
        process["listening"] = True
        self.processes[self.id] = process
        print "Process %s now listening!(%s)"%(self.id, process)
        while inputs:
            readable, writable, exceptional = select.select(inputs,
                                                           outputs,
                                                           inputs,
                                                           0.1)
            for sock in readable:
                print "Process %s has a readable scoket: %s"%(self.id,
                                                              sock)
                if sock is server:
                    print "Process %s has a readable server scoket: %s"%(self.id,
                                                              sock)
                    conn, addr = sock.accept()
                    conn.setblocking(0)
                    inputs.append(conn)
                else:
                    data = sock.recv(1024)
                    if data:
                        self.queue.append(data)
                        print "non server readable socket with data"
                    else:
                        inputs.remove(sock)
                        sock.close()
                        print "non server readable socket with no data"

            for sock in exceptional:
                print "exception occured on socket %s"%(sock)
                inputs.remove(sock)
                sock.close()

            while len(self.queue) >= 1:
                print "Received:", self.queue.pop()

            # send a message to a random process:
            random_id = random.choice(list(self.processes.keys()))
            print "%s Attempting to send message to %s"%(self.id, random_id)
            random_process = self.processes[random_id]
            print "random_process:", random_process
            if random_process["listening"]:
                random_address = random_process["address"]
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                try:
                    s.connect(random_address)
                except socket.error:
                    print "%s failed to send to %s"%(self.id, random_id)
                else:
                    s.send("Hello World!")                    
                finally:
                    s.close()

            time.sleep(1)

if __name__=="__main__":
    print "hostname:", socket.getfqdn()
    print dir(multiprocessing)
    manager = multiprocessing.Manager()
    processes = manager.dict()
    joinable = []
    for n in xrange(multiprocessing.cpu_count()):
        mpp = MessagePassingProcess(n, processes)
        processes[n] = {"id":n,
                        "address":("127.0.0.1",7000+n),
                        "listening":False,
                        }
        print "processes[%s] = %s"%(n, processes[n])
        mpp.start()
        joinable.append(mpp)
    for process in joinable:
        process.join()

经过大量润色和测试，这可能是对 multiprocessing.Process 和/或 multiprocessing.Pool 的合理扩展，因为这看起来确实是人们会使用的东西如果它在标准库中可用。创建一个 DynamicQueue 类也可能是合理的，该类使用可被其他队列发现的套接字。

无论如何，希望对您有所帮助。如果您找到更好的方法来完成这项工作，请更新。

关于python - 如何在 Python 多处理中动态创建每个进程队列，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/7072875/