进程运行时如何在它们之间进行通信 [英] How to communicate between processes while they're running

问题描述

我想有两个并行运行的进程.一个从另一个获取输入，处理数据，并将处理后的数据作为另一个输出发送回去.另一个过程做同样的事情.显然，需要有一个起点和终点.

I would like to have two processes which run parallel. The one gets input from the other, processes the data and sends the processed data back as output for the other. The other process does the same thing. Obviously there needs to be a starting point and an end point.

当进程正在运行时，如何在它们之间进行通信?我只设法彼此依次运行这两个进程.

How can I communicate between the processes while they're running? I only managed to run the two processes after each other.

我试图用multiprocessing解决它:

from multiprocessing import Process, Queue, Array
sentinel = -1

def process1(q, arr):
    # Receives data, modifies it and sends it back
    while True:
        data = q.get() # Receive data

        if data is sentinel:
            break
    data *= 2 # Data modification
    arr.append(data) # Data logging
    q.put(data) # Send data
    q.put(sentinel) # Send end signal

def process2(q, arr):
    # Receives data, increments it and sends data back
    if q.empty():
        data = 1
    else:
        while True:
            data = q.get() # Receive data
            if data == sentinel:
                break

    data += 1
    q.put(data) # Send data
    arr.append(data) # Data logging
    q.put(sentinel) # Send end signal

if __name__ == "__main__":
    q = Queue()
    logbook = Array('i', 0)
    counter = 0
    while counter < 10:
        process_1 = Process(target=process1, args=(q, logbook))
        process_2 = Process(target=process2, args=(q, logbook))
        process_1.start()
        process_2.start()
        q.close()
        q.join_thread()
        process_1.join()
        process_2.join()
        counter += 1
    print(logbook)

推荐答案

我试图了解您的需求，但是对我来说还不是很清楚，因此我建议使用此生产者-消费者版本的代码，两个进程可以在该版本中进行通信在一定数量的迭代中达到最终结果.

I tried to understand your need, but it is not fully clear to me, thus I propose this producer-consumer version of the code, where the two process communicate to reach the final result for a certain amount of iterations.

首先，您需要两个队列，以避免将内容放入队列的同一过程先读取另​​一个队列. 其次，您需要一种在计算结束时达成一致的机制，在这种情况下为None消息.

First of all you need two queues in order to avoid that the same process that puts the content into the queue reads it before the other one. Second, you need a mechanism to agree on the end of the computation, in this case a None message.

下面的代码总结了我提出的解决方案:

My proposed solution is summarised in the following code:

from multiprocessing import Process, Queue, Array

def process1(in_queue, out_queue):
    # Receives data, modifies it and sends it back
    while True:
        data = in_queue.get() # Receive data
        if data is None:
            out_queue.put(None)  # send feedback about END message
            out_queue.close()
            out_queue.join_thread()
            break

        data *= 2 # Data modification
        out_queue.put(data) # Send data

def process2(in_queue, out_queue, how_many):
    data = 0

    # Receives data, increments it and sends data back
    while how_many > 0:
        data += 1 # Data modification
        out_queue.put(data) # Send data
        how_many -= 1

        data = in_queue.get() # Receive data
        if data is None:
            break

    # send END message
    out_queue.put(None)
    out_queue.close()
    out_queue.join_thread()
    assert in_queue.get() is None


if __name__ == "__main__":
    q1 = Queue()
    q2 = Queue()

    process_1 = Process(target=process1, args=(q1, q2))
    process_2 = Process(target=process2, args=(q2, q1, 10))
    process_1.start()
    process_2.start()

    process_2.join()
    process_1.join()
    q1.close()
    q2.close()
    q1.join_thread()
    q2.join_thread()

这篇关于进程运行时如何在它们之间进行通信的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！