我正在制作一个项目,从客户端传感器收集数据,处理收集到的数据并将其发送回客户端。可能有多个客户端要求同时从我们的服务器接收一些数据,所以我必须实现多重处理。我无法使用线程,因为某些变量必须与客户端无关。如果我这样做,我的代码可能会变得非常难以阅读和升级,而我不希望这样。所以我决定使用Processes,但是现在有一些数据需要在父进程和子进程之间进行剪切。经过一番研究,我发现Pipe通信可以满足我的要求。
以下代码成功将数据从父进程发送到子进程,子进程更新数据并将其发送回父进程。但它之所以起作用只是因为 sleep() 函数阻止父级与子级同时使用管道。
如何更改它以使其执行相同的操作,但没有 sleep() 函数,我相信它很可能会在将来引起问题?
from multiprocessing import Process, Pipe
import time
def update_data(pipe):
p_out, p_in = pipe
L = []
while True:
message = p_out.recv()
if message=='FINISHED':
break
L.append(message)
L.append(['new data']) #updating received data
writer(L, p_in) #sending received data to parent Process
p_in.close()
def writer(i, p_in):
p_in.send(i)
p_in.send('FINISHED')
L = ['0' for i in range(10)] #current data
if __name__=='__main__':
p_out, p_in = Pipe()
update_data_process = Process(target=update_data, args=((p_out, p_in),))
update_data_process.start()
writer(L, p_in) #sending current data to child Process
time.sleep(3) #needs to be changed
while True:
message = p_out.recv()
if message != 'FINISHED':
L = message
else:
break
print(L)
p_in.close()
update_data_process.join()
最佳答案
您遇到这个问题是因为您将连接视为单工,但 Pipe() 默认情况下返回双工(双向)连接。
这意味着当您调用 parent_conn, child_conn = Pipe()
时,您将获得一个连接,仅父级应该用于读取并且 为子级写入和另一个这样的连接对象。父级和子级仅对其连接对象进行操作。
from multiprocessing import Process, Pipe
from datetime import datetime
SENTINEL = 'SENTINEL'
def update_data(child_conn):
result = []
for msg in iter(child_conn.recv, SENTINEL):
print(f'{datetime.now()} child received {msg}')
result.append(msg)
print(f'{datetime.now()} child received sentinel')
result.append(['new data'])
writer(child_conn, result)
child_conn.close()
def writer(conn, data):
conn.send(data)
conn.send(SENTINEL)
if __name__=='__main__':
parent_conn, child_conn = Pipe() # default is duplex!
update_data_process = Process(target=update_data, args=(child_conn,))
update_data_process.start()
data = ['0' for i in range(3)]
writer(parent_conn, data)
for msg in iter(parent_conn.recv, SENTINEL):
print(f'{datetime.now()} parent received {msg}')
print(f'{datetime.now()} parent received sentinel')
parent_conn.close()
update_data_process.join()
示例输出:
2019-03-12 00:09:52.920375 child received ['0', '0', '0']
2019-03-12 00:09:52.920512 child received sentinel
2019-03-12 00:09:52.920702 parent received [['0', '0', '0'], ['new data']]
2019-03-12 00:09:52.920764 parent received sentinel
Process finished with exit code 0
如果您不熟悉 iter(callable, Sentinel) 的使用
,请查看 here 。
关于Python 多处理 - 进程之间的管道通信,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55110733/