使用多处理的Python进程之间共享数据的问题

Question

我已经看过几篇关于这个的帖子，所以我知道这是相当简单的，但我似乎做得不够。我不确定是需要创建工作池还是使用Queue类。基本上，我希望能够创建多个进程，每个进程都是自主的（这就是它们从代理超类继承的原因）。在

在主循环的随机滴答声中，我想更新每个代理。我在主循环和代理的运行循环中使用具有不同值的time.sleep来模拟不同的处理器速度。在

这是我的代理超类：

# Generic class to handle mpc of each agent
class Agent(mpc.Process):
  # initialize agent parameters
  def __init__(self,):
    # init mpc
    mpc.Process.__init__(self)
    self.exit = mpc.Event()

  # an agent's main loop...generally should be overridden
  def run(self):
    while not self.exit.is_set():
      pass
    print "You exited!"

  # safely shutdown an agent
  def shutdown(self):
    print "Shutdown initiated"
    self.exit.set()

  # safely communicate values to this agent
  def communicate(self,value):
    print value

特定代理的子类（模拟HVAC系统）：

^{pr2}$

我的主循环：

if __name__ == "__main__":
  print "Initializing subsystems"
  agents = {}
  agents['HVAC'] = HVAC()

  # Run simulation
  timestep = 0
  while timestep < args.timesteps:
    print "Timestep %d" % timestep

    if timestep % 10 == 0:
      curr_temp = random.randrange(68,72)
      curr_humidity = random.uniform(40.0,60.0)
      agents['HVAC'].communicate({'temp':curr_temp, 'humidity':curr_humidity})

    time.sleep(1)
    timestep += 1

  agents['HVAC'].shutdown()
  print "HVAC process state: %d" % agents['HVAC'].is_alive()

所以问题是，每当我在主循环中运行agents['HVAC'].communicate(x)时，我都可以看到值被传递到其run循环中的HVAC子类中（因此它正确地打印接收到的值）。但是，该值从未成功存储。在

所以典型的输出如下：

Initializing subsystems
Timestep 0
Measured [68] [56.948675]
heating 1
heating 2
Timestep 1
heating 3
heating 4
Timestep 2
heating 5
heating 6

实际上，一旦出现测量值[68]，内部存储值应更新为输出68（不是加热1、加热2等）。如此有效地，暖通空调的自身温度没有得到适当的更新。在

---

编辑：经过一番研究，我意识到我并不一定了解幕后发生了什么。每个子进程都使用自己的虚拟内存块来运行，并且完全从这种方式共享的任何数据中抽象出来，因此传入值是行不通的。我的新问题是，我不一定确定如何与多个进程共享全局值。在

我正在查看Queue或JoinableQueue包，但我不一定确定如何将队列传递到我拥有的超类设置类型中（尤其是使用mpc.Process.__init__(self)调用）。在

另一个问题是，我是否可以让多个代理从队列中读取值而不将其从队列中拉出？例如，如果我想与多个代理共享一个temperature值，那么一个队列是否可以用于此目的？在

Pipe v Queue

Answer 1

以下是一个建议的解决方案，假设您希望：

• 控制工人寿命的集中管理者/主要过程
• 工作进程独立完成某项工作，然后将结果报告给经理和其他进程

在我展示之前，我想说的是，一般情况下，除非您是CPU专用的，否则{}实际上并不适合，主要是因为增加了复杂性，而且您最好使用不同的高级异步框架。另外，你应该使用python3，它好得多！在

也就是说，^{}，使用multiprocessing可以很容易地实现这一点。我已经在Python3中完成了这项工作，但我不认为任何东西在Python2中不应该“只工作”，但我还没有检查过。在

from ctypes import c_bool
from multiprocessing import Manager, Process, Array, Value
from pprint import pprint
from time import sleep, time


class Agent(Process):

    def __init__(self, name, shared_dictionary, delay=0.5):
        """My take on your Agent.

        Key difference is that I've commonized the run-loop and used
        a shared value to signal when to stop, to demonstrate it.
        """
        super(Agent, self).__init__()
        self.name = name

        # This is going to be how we communicate between processes.
        self.shared_dictionary = shared_dictionary

        # Create a silo for us to use.
        shared_dictionary[name] = []
        self.should_stop = Value(c_bool, False)

        # Primarily for testing purposes, and for simulating 
        # slower agents.
        self.delay = delay

    def get_next_results(self):
        # In the real world I'd use abc.ABCMeta as the metaclass to do 
        # this properly.
        raise RuntimeError('Subclasses must implement this')

    def run(self):
        ii = 0
        while not self.should_stop.value:
            ii += 1
            # debugging / monitoring
            print('%s %s run loop execution %d' % (
                type(self).__name__, self.name, ii))

            next_results = self.get_next_results()

            # Add the results, along with a timestamp.
            self.shared_dictionary[self.name] += [(time(), next_results)]
            sleep(self.delay)

    def stop(self):
        self.should_stop.value = True
        print('%s %s stopped' % (type(self).__name__, self.name))


class HVACAgent(Agent):
    def get_next_results(self):
        # This is where you do your work, but for the sake of
        # the example just return a constant dictionary.
        return {'temperature': 5, 'pressure': 7, 'humidity': 9}


class DumbReadingAgent(Agent):
    """A dumb agent to demonstrate workers reading other worker values."""

    def get_next_results(self):
        # get hvac 1 results:
        hvac1_results = self.shared_dictionary.get('hvac 1')
        if hvac1_results is None:
            return None

        return hvac1_results[-1][1]['temperature']

# Script starts.
results = {}

# The "with" ensures we terminate the manager at the end.
with Manager() as manager:

    # the manager is a subprocess in its own right. We can ask
    # it to manage a dictionary (or other python types) for us
    # to be shared among the other children.
    shared_info = manager.dict()

    hvac_agent1 = HVACAgent('hvac 1', shared_info)
    hvac_agent2 = HVACAgent('hvac 2', shared_info, delay=0.1)
    dumb_agent = DumbReadingAgent('dumb hvac1 reader', shared_info)

    agents = (hvac_agent1, hvac_agent2, dumb_agent)

    list(map(lambda a: a.start(), agents))

    sleep(1)

    list(map(lambda a: a.stop(), agents))
    list(map(lambda a: a.join(), agents))

    # Not quite sure what happens to the shared dictionary after
    # the manager dies, so for safety make a local copy.
    results = dict(shared_info)

pprint(results)