1

我有一组进程，我们称它们为 A、B 和 C，它们需要相互通信。A 需要与 B 和 C 通信；B 需要与 A 和 C 通信；C 需要与 A 和 B 通信。A、B 和 C 可以位于不同的机器上，也可以位于同一台机器上。

我的想法是通过套接字进行通信，如果它们都在同一台机器上，则使用“localhost”（例如，A 在端口 11111，B 在端口 22222 等）。这样，非本地进程将被视为本地进程。为此，我想我会为 A、B 和 C 中的每一个设置一个 SocketServer 实例，并且它们中的每一个都会知道另外两个的地址。每当需要进行通信时，例如 A 到 B，A 就会打开一个到 B 的套接字并写入数据。然后 B 的持续运行的服务器将读取数据并将其存储在列表中以供以后需要时使用。

我遇到的问题是存储的信息没有在finish_request方法（处理监听）和__call__方法（处理谈话）之间共享。（服务器类是可调用的，因为我需要它来做其他事情。我认为这与问题无关。）

我的问题是这会像我想象的那样工作吗？将multiprocessing,threading和socketserver一起在同一台机器上玩得好吗？我对使用其他机制在进程之间进行通信（如Queueor Pipe）不感兴趣。我有一个可行的解决方案。我想知道这种方法是否可行，即使效率较低。而且，如果是的话，我做错了什么阻止它工作？

下面是说明该问题的最小示例：

import uuid
import sys
import socket
import time
import threading
import collections
import SocketServer
import multiprocessing

class NetworkMigrator(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
    def __init__(self, server_address, client_addresses, max_migrants=1):
        SocketServer.TCPServer.__init__(self, server_address, None)
        self.client_addresses = client_addresses
        self.migrants = collections.deque(maxlen=max_migrants)
        self.allow_reuse_address = True
        t = threading.Thread(target=self.serve_forever)
        t.daemon = True
        t.start()

    def finish_request(self, request, client_address):
        try:
            rbufsize = -1
            wbufsize = 0
            rfile = request.makefile('rb', rbufsize)
            wfile = request.makefile('wb', wbufsize)

            data = rfile.readline().strip()
            self.migrants.append(data)
            print("finish_request::  From: %d  To: %d  MID: %d  Size: %d -- %s" % (client_address[1], 
                                                                                   self.server_address[1], 
                                                                                   id(self.migrants), 
                                                                                   len(self.migrants), 
                                                                                   data))

            if not wfile.closed:
                wfile.flush()
            wfile.close()
            rfile.close()        
        finally:
            sys.exc_traceback = None

    def __call__(self, random, population, args):
        client_address = random.choice(self.client_addresses)
        migrant_index = random.randint(0, len(population) - 1)
        data = population[migrant_index]
        data = uuid.uuid4().hex
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            sock.connect(client_address)
            sock.send(data + '\n')
        finally:
            sock.close()
        print("      __call__::  From: %d  To: %d  MID: %d  Size: %d -- %s" % (self.server_address[1], 
                                                                               client_address[1], 
                                                                               id(self.migrants), 
                                                                               len(self.migrants), 
                                                                               data))
        if len(self.migrants) > 0:
            migrant = self.migrants.popleft()
            population[migrant_index] = migrant
        return population


def run_it(migrator, rand, pop):
    for i in range(10):
        pop = migrator(r, pop, {})
        print("        run_it::  Port: %d  MID: %d  Size: %d" % (migrator.server_address[1], 
                                                                 id(migrator.migrants), 
                                                                 len(migrator.migrants)))
        time.sleep(1)


if __name__ == '__main__':
    import random
    r = random.Random()
    a = ('localhost', 11111)
    b = ('localhost', 22222)
    c = ('localhost', 33333)
    am = NetworkMigrator(a, [b, c], max_migrants=11)
    bm = NetworkMigrator(b, [a, c], max_migrants=22)
    cm = NetworkMigrator(c, [a, b], max_migrants=33)

    fun = [am, bm, cm]
    pop = [["larry", "moe", "curly"], ["red", "green", "blue"], ["small", "medium", "large"]]
    jobs = []
    for f, p in zip(fun, pop):
        pro = multiprocessing.Process(target=run_it, args=(f, r, p))
        jobs.append(pro)
        pro.start()
    for j in jobs:
        j.join()
    am.shutdown()
    bm.shutdown()
    cm.shutdown()

查看此示例的输出，将有三种类型的打印： 

        run_it::  Port: 11111  MID: 3071227860  Size: 0
      __call__::  From: 11111  To: 22222  MID: 3071227860  Size: 0 -- e00e0891e0714f99b86e9ad743731a00
finish_request::  From: 60782  To: 22222  MID: 3071227972  Size: 10 -- e00e0891e0714f99b86e9ad743731a00

“MID”是该实例中的id如果双端队列。migrants“From”和“To”是发送/接收传输的端口。我现在只是将数据设置为随机的十六进制字符串，以便我可以跟踪单个传输。

我不明白为什么，即使使用相同的 MID，有时它会说它的大小是非零的，然后它会说它的大小是 0。我觉得它必须源于这样一个事实调用是多线程的。如果使用这些行而不是最后的 2 个for循环，系统将按我预期的方式工作：

for _ in range(10):
    for f, p in zip(fun, pop):
        f(r, p, {})
        time.sleep(1)

那么破坏它的多处理版本发生了什么？

4

1 回答 1

2

当我们创建 3 个新的 NetworkMigrator 对象时，会启动 3 个新线程，每个线程都在侦听新的 TCP 连接。稍后，我们为 run_it 函数启动 3 个新进程。我们总共有 4 个进程，第一个进程包含 4 个线程（1 个主线程 + 3 个服务器）。现在，问题是其他 3 个进程将无法访问侦听服务器线程对对象所做的更改。这是因为默认情况下进程不共享内存。

因此，如果您启动 3 个新线程而不是进程，您会注意到不同之处：

pro = threading.Thread(target=run_it,args=(f,r,p))

还有一个小问题。线程之间的这种共享也不是完全安全的。每当我们更改对象的状态时，最好使用锁。最好在 finish_request 和call方法中执行类似下面的操作。

lock = Lock()
...
lock.acquire()    
self.migrants.append(data)
lock.release()

如果您对多线程不满意并且确实想要多处理，那么您可以使用代理对象，如下所述：http: //docs.python.org/library/multiprocessing.html#proxy-objects

至于对象ID相同，这并不意外。新进程在该时间点传递对象的状态（包括对象 ID）。新进程继续保留这些对象 ID，但我们在这里讨论的是两个完全不同的内存空间，因为它们是不同的进程。因此，主流程所做的任何更改都不会反映在创建的子流程中。

于 2012-02-23T04:57:12.990 回答