python - 如何将python dict与多处理同步

Question

我正在使用 Python 2.6 和用于多线程的多处理模块。现在我想要一个同步的字典(我真正需要的唯一原子操作是值上的 += 运算符)。

我应该用 multiprocessing.sharedctypes.synchronized() 调用包装 dict 吗？或者是另一种方式？

Answer 1

介绍

似乎有很多扶手椅建议，但没有工作示例。这里列出的答案甚至都没有建议使用多处理，这有点令人失望和不安。作为 Python 爱好者，我们应该支持我们的内置库，虽然并行处理和同步从来都不是一件小事，但我相信通过适当的设计，它可以变得微不足道。这在现代多核架构中变得极其重要，怎么强调都不过分!也就是说，我对 multiprocessing 库并不满意，因为它仍处于起步阶段，有很多陷阱、错误，并且面向函数式编程(我讨厌它)。目前我还是更喜欢Pyro由于多处理的严重限制无法在服务器运行时共享新创建的对象，因此模块(这远远领先于它的时间)超过了多处理。管理器对象的“注册”类方法只会在管理器(或其服务器)启动之前实际注册一个对象。废话不多说，再来点代码:

服务器.py

from multiprocessing.managers import SyncManager


class MyManager(SyncManager):
    pass


syncdict = {}
def get_dict():
    return syncdict

if __name__ == "__main__":
    MyManager.register("syncdict", get_dict)
    manager = MyManager(("127.0.0.1", 5000), authkey="password")
    manager.start()
    raw_input("Press any key to kill server".center(50, "-"))
    manager.shutdown()

在上面的代码示例中，Server.py 使用了可以提供同步共享对象的 multiprocessing 的 SyncManager。这段代码在解释器中无法运行，因为多处理库对于如何为每个注册的对象查找“可调用对象”非常敏感。运行 Server.py 将启动一个定制的 SyncManager，它共享同步字典以供多个进程使用，并且可以连接到同一台机器上的客户端，或者如果在环回以外的 IP 地址上运行，则可以连接到其他机器。在这种情况下，服务器在端口 5000 上的环回 (127.0.0.1) 上运行。使用 authkey 参数在操作同步字典时使用安全连接。当按下任何键时，管理器将关闭。

客户端.py

from multiprocessing.managers import SyncManager
import sys, time

class MyManager(SyncManager):
    pass

MyManager.register("syncdict")

if __name__ == "__main__":
    manager = MyManager(("127.0.0.1", 5000), authkey="password")
    manager.connect()
    syncdict = manager.syncdict()

    print "dict = %s" % (dir(syncdict))
    key = raw_input("Enter key to update: ")
    inc = float(raw_input("Enter increment: "))
    sleep = float(raw_input("Enter sleep time (sec): "))

    try:
         #if the key doesn't exist create it
         if not syncdict.has_key(key):
             syncdict.update([(key, 0)])
         #increment key value every sleep seconds
         #then print syncdict
         while True:
              syncdict.update([(key, syncdict.get(key) + inc)])
              time.sleep(sleep)
              print "%s" % (syncdict)
    except KeyboardInterrupt:
         print "Killed client"

客户端还必须创建一个定制的 SyncManager，注册“syncdict”，这次没有传入一个可调用来检索共享 dict。然后，它使用自定义的 SycnManager 使用端口 5000 上的环回 IP 地址 (127.0.0.1) 和在 Server.py 中启动的与管理器建立安全连接的 authkey 进行连接。它通过调用管理器上注册的可调用对象来检索共享字典同步字典。它会提示用户进行以下操作:

syncdict 中操作的键

每个周期增加键访问的值的数量

每个周期的 sleep 时间(以秒为单位)

然后客户端检查 key 是否存在。如果不是，它会在同步字典上创建 key 。客户端然后进入一个“无限”循环，它通过增量更新键的值，休眠指定的数量，并仅打印同步字典以重复此过程，直到发生 KeyboardInterrupt (Ctrl+C)。

烦人的问题

管理器的注册方法必须在管理器启动之前调用，否则即使对管理器的 dir 调用将显示它确实具有已注册的方法，您也会得到异常。

dict 的所有操作都必须使用方法而不是 dict 赋值来完成(由于 multiprocessing 共享自定义对象的方式，syncdict["blast"] = 2 将失败)

使用 SyncManager 的 dict 方法将缓解烦人的问题 #2，除了烦人的问题 #1 会阻止 SyncManager.dict() 返回的代理被注册和共享。 (SyncManager.dict() 只能在管理器启动后调用，并且 register 只会在管理器启动前工作，所以 SyncManager.dict() 仅在进行函数式编程并将代理作为参数传递给 Processes 时才有用，例如文档示例)

服务器和客户端都必须注册，即使直觉上看起来客户端在连接到管理器后才能弄清楚(请将此添加到您的愿望 list 多处理开发人员)

关闭

我希望你和我一样喜欢这个非常彻底且有点耗时的答案。我在想清楚为什么我在 Pyro 让它轻而易举的多处理模块上如此挣扎时遇到了很多麻烦，现在多亏了这个答案，我才一针见血。我希望这对 python 社区如何改进多处理模块有用，因为我相信它有很大的希望，但在它的初期还没有达到可能。尽管描述了烦人的问题，但我认为这仍然是一个非常可行的替代方案，而且非常简单。您还可以使用 SyncManager.dict() 并将其作为文档显示的方式传递给 Processes ，它可能是一个更简单的解决方案，具体取决于您的要求，这对我来说是不自然的。