回答此问题可获得 20 贡献值,回答如果被采纳可获得 50 分。
<p><strong>问题</strong></p>
<p>有两个独立的进程并行运行,我希望它们能够来回通信</p>
<p><strong>代码说明</strong></p>
<p>代码在Python2.7中。在我的精简脚本中,我使用队列进行进程间通信。进程p1将数据放入队列中。进程p2从队列中获取数据,并对数据进行处理。然后进程p2将修改后的数据放回队列,最后进程p1从队列中取回修改后的数据。修改后的数据必须返回到进程p1,因为该进程实际上是一个发送/接收请求的eventlet服务器</p>
<p><strong>代码</strong></p>
<pre><code>#!/usr/bin/python2.7 python2.7
# -*- coding: utf-8 -*-
# script for back-and-forth data exchange between processes
# common modules
import os
import sys
import time
from multiprocessing import Process
from multiprocessing import Queue
from datetime import datetime
someData = {}
class Load():
def post(self):
timestamp = str(datetime.now())
someData = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
queue1.put(someData) # put into queue
print "#20 process 1: put in queue1 =>", someData
time.sleep(3)
while True: # queue1 checking loop, comment out the loop if use time.sleep only
if queue1.empty() == False:
timestamp = str(datetime.now())
res = queue1.get()
res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
print "#28 get from queue1 =>", res
break
else:
print "#31 queue1 empty"
time.sleep(1)
# while True: # queue2 checking loop
# if queue2.empty() == False:
# timestamp = str(datetime.now())
# res = queue2.get()
# res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
# print "#39 get from queue2 =>", res
# break
# else:
# print "#42 queue2 empty"
# time.sleep(1)
class Unload():
def get(self):
try:
if queue1.empty() == False:
data = queue1.get() # retrieve package from queue
#queue1.close()
#queue1.join_thread()
timestamp = str(datetime.now())
data = {"process":"p2","class":"Unload()","method":"get()","timestamp":timestamp}
print "#54 process 2: get from queue1 =>", data
self.doSomething(data) # call method
else:
print "#57 queue1 empty"
pass
except:
print "#60 queue1 error"
pass
def doSomething(self, data):
time.sleep(3)
timestamp = str(datetime.now())
someData = {"process":"p2","class":"Unload()","method":"doSomething()","timestamp":timestamp}
self.someData = someData
print "#68 process 2: do something =>", someData
self.put()
def put(self):
time.sleep(3)
timestamp = str(datetime.now())
self.someData = {"process":"p2","class":"Unload()","method":"put()","timestamp":timestamp}
print "#75 process 2: put back in queue1 =>", self.someData
res = self.someData
queue1.put(res)
#print "#78 process 2: put back in queue2 =>", self.someData
#res = self.someData
#queue2.put(res)
#queue2.close()
#queue2.join_thread()
# main
if __name__ == '__main__':
queue1 = Queue()
#queue2 = Queue()
global p1, p2
p1 = Process(target=Load().post(), args=(queue1,)) # process p1
#p1 = Process(target=Load().post(), args=(queue1,queue2,))
p1.daemon = True
p1.start()
p2 = Process(target=Unload().get(), args=(queue1,)) # process p2
#p2 = Process(target=Unload().get(), args=(queue1,queue2,))
p2.start()
p2.join()
</code></pre>
<p><strong>问题</strong>
我已经检查了这方面的其他资源,但它们都涉及单向沟通。下面是资源列表</p>
<ol>
<li><a href="https://stackoverflow.com/questions/39170249/use-get-nowait-in-python-without-raising-empty-exception">use-get-nowait-in-python-without-raising-empty-exception</a></li>
<li><a href="https://stackoverflow.com/questions/11216702/in-python-how-do-you-get-data-back-from-a-particular-process-using-multiprocess">in-python-how-do-you-get-data-back-from-a-particular-process-using-multiprocess</a></li>
<li><a href="https://stackoverflow.com/questions/57095895/how-to-use-multiprocessing-queue-with-lock">how-to-use-multiprocessing-queue-with-lock</a></li>
<li><a href="https://www.blog.pythonlibrary.org/2016/08/02/python-201-a-multiprocessing-tutorial/" rel="nofollow noreferrer">multiprocessing module supports locks</a></li>
<li><a href="https://stackoverflow.com/questions/33640283/thread-that-i-can-pause-and-resume">thread-that-i-can-pause-and-resume</a></li>
<li><a href="https://stackoverflow.com/questions/45066782/exchange-data-between-two-python-processes">exchange-data-between-two-python-processes</a></li>
</ol>
<p>如何让process1等待并从process2检索修改后的数据?我应该考虑另一种方法来处理进程之间的通信吗?例如,管道,ZNOMQ?<p>
<p><strong>尝试1:在进程1中不使用while循环而使用time.sleep()
仅使用time.sleep,数据在队列中上下移动,但永远不会到达进程1中的最终目的地。到目前为止一切都很好,但最后一步还没有完成。结果如下</p>
<pre><code>#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:40:30.234466', 'class': 'Load()', 'method': 'post()'}
#54 process 2: get from queue1 => {'process': 'p2', 'timestamp': '2020-02-23 11:40:33.239113', 'class': 'Unload()', 'method': 'get()'}
#68 process 2: do something => {'process': 'p2', 'timestamp': '2020-02-23 11:40:36.242500', 'class': 'Unload()', 'method': 'doSomething()'}
#75 process 2: put back in queue1 => {'process': 'p2', 'timestamp': '2020-02-23 11:40:39.245856', 'class': 'Unload()', 'method': 'put()'}
</code></pre>
<p><strong>尝试2:在进程1中使用while循环
通过while循环检查队列,数据进入队列后立即被捕获,它们永远不会到达进程2。结果如下</p>
<pre><code>#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:46:14.606356', 'class': 'Load()', 'method': 'post()'}
#28 get from queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:46:17.610202', 'class': 'Load()', 'method': 'post()'}
#57 queue1 empty
</code></pre>
<p><strong>尝试3:使用两个队列
使用两个队列:队列1从process1到process2,队列2从process2到process1。数据进入队列1,但不返回队列2,它们神秘地消失了。结果如下</p>
<pre><code>#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:53:39.745177', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty
</code></pre>
<p><strong>----更新20200224:尝试4、5和6---------------------------------------------------------------------</strong></p>
<p><strong>尝试4:使用manager.Queue()的两个队列</strong></p>
<p>在manager.Queue()中使用两个队列:队列1从process1到process2,队列2从process2到process1。数据进入队列1,但不返回队列2,它们再次神秘地消失。代码和结果如下所示</p>
<p>尝试4的代码:
#!/usr/bin/python2.7 python2.7
#编码:utf-8-</em>-
#用于序列化进程间数据交换的脚本</p>
<pre><code># common modules
import os
import sys
import time
import multiprocessing
from multiprocessing import Process
from multiprocessing import Queue
from multiprocessing import Manager
from datetime import datetime
someData = {}
manager = multiprocessing.Manager()
queue1 = manager.Queue()
queue2 = manager.Queue()
class Load():
def post(self):
timestamp = str(datetime.now())
someData = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
queue1.put(someData) # put into queue
print "#20 process 1: put in queue1 =>", someData
time.sleep(3)
# while True: # queue1 checking loop
# if queue1.empty() == False:
# timestamp = str(datetime.now())
# res = queue1.get()
# res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
# print "#28 get from queue1 =>", res
# break
# else:
# print "#31 queue1 empty"
# time.sleep(1)
while True: # queue2 checking loop
if queue2.empty() == False:
timestamp = str(datetime.now())
res = queue2.get()
res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
print "#39 get from queue2 =>", res
break
else:
print "#42 queue2 empty"
time.sleep(1)
class Unload():
def get(self):
try:
if queue1.empty() == False:
data = queue1.get() # retrieve package from queue
#queue1.close()
#queue1.join_thread()
timestamp = str(datetime.now())
data = {"process":"p2","class":"Unload()","method":"get()","timestamp":timestamp}
print "#54 process 2: get from queue1 =>", data
self.doSomething(data) # call method
else:
print "#57 queue1 empty"
pass
except:
print "#60 queue1 error"
pass
def doSomething(self, data):
time.sleep(3)
timestamp = str(datetime.now())
someData = {"process":"p2","class":"Unload()","method":"doSomething()","timestamp":timestamp}
self.someData = someData
print "#68 process 2: do something =>", someData
self.put()
def put(self):
time.sleep(3)
timestamp = str(datetime.now())
self.someData = {"process":"p2","class":"Unload()","method":"put()","timestamp":timestamp}
res = self.someData
#print "#75 process 2: put back in queue1 =>", self.someData
#queue1.put(res)
print "#78 process 2: put back in queue2 =>", self.someData
queue2.put(res)
#queue2.close()
#queue2.join_thread()
# main
if __name__ == '__main__':
manager = multiprocessing.Manager()
queue1 = manager.Queue()
queue2 = manager.Queue()
global p1, p2
#p1 = Process(target=Load().post(), args=(queue1,)) # process p1
p1 = Process(target=Load().post(), args=(queue1,queue2,))
p1.daemon = True
p1.start()
#p2 = Process(target=Unload().get(), args=(queue1,)) # process p2
p2 = Process(target=Unload().get(), args=(queue1,queue2,))
p2.start()
p2.join()
</code></pre>
<p>尝试4的结果:</p>
<pre><code>#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 13:06:17.687762', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty
</code></pre>
<p><strong>尝试5:与manager.queue()一起使用一个队列</strong>
在manager.queue()中使用一个队列:queue1从process1到process2,queue1从process2返回到process1。数据进入队列1,但随后立即被捕获,它们永远不会到达进程2。代码结果如下所示</p>
<p>尝试5的代码:</p>
<pre><code>#!/usr/bin/python2.7 python2.7
# -*- coding: utf-8 -*-
# script for serialized interprocess data exchange
# common modules
import os
import sys
import time
import multiprocessing
from multiprocessing import Process
from multiprocessing import Queue
from multiprocessing import Manager
from datetime import datetime
someData = {}
manager = multiprocessing.Manager()
queue1 = manager.Queue()
#queue2 = manager.Queue()
class Load():
def post(self):
timestamp = str(datetime.now())
someData = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
queue1.put(someData) # put into queue
print "#25 process 1: put in queue1 =>", someData
time.sleep(3)
while True: # queue1 checking loop
if queue1.empty() == False:
timestamp = str(datetime.now())
res = queue1.get()
res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
print "#33 get from queue1 =>", res
break
else:
print "#36 queue1 empty"
time.sleep(1)
# while True: # queue2 checking loop
# if queue2.empty() == False:
# timestamp = str(datetime.now())
# res = queue2.get()
# res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
# print "#44 get from queue2 =>", res
# break
# else:
# print "#47 queue2 empty"
# time.sleep(1)
class Unload():
def get(self):
try:
if queue1.empty() == False:
data = queue1.get() # retrieve package from queue
#queue1.close()
#queue1.join_thread()
timestamp = str(datetime.now())
data = {"process":"p2","class":"Unload()","method":"get()","timestamp":timestamp}
print "#59 process 2: get from queue1 =>", data
self.doSomething(data) # call method
else:
print "#62 queue1 empty"
pass
except:
print "#65 queue1 error"
pass
def doSomething(self, data):
time.sleep(3)
timestamp = str(datetime.now())
someData = {"process":"p2","class":"Unload()","method":"doSomething()","timestamp":timestamp}
self.someData = someData
print "#73 process 2: do something =>", someData
self.put()
def put(self):
time.sleep(3)
timestamp = str(datetime.now())
self.someData = {"process":"p2","class":"Unload()","method":"put()","timestamp":timestamp}
res = self.someData
print "#81 process 2: put back in queue1 =>", self.someData
queue1.put(res)
#print "#83 process 2: put back in queue2 =>", self.someData
#queue2.put(res)
#queue2.close()
#queue2.join_thread()
# main
if __name__ == '__main__':
manager = multiprocessing.Manager()
queue1 = manager.Queue()
#queue2 = manager.Queue()
global p1, p2
p1 = Process(target=Load().post(), args=(queue1,)) # process p1
#p1 = Process(target=Load().post(), args=(queue1,queue2,))
p1.daemon = True
p1.start()
p2 = Process(target=Unload().get(), args=(queue1,)) # process p2
#p2 = Process(target=Unload().get(), args=(queue1,queue2,))
p2.start()
p2.join()
</code></pre>
<p>尝试5的结果:</p>
<pre><code>#25 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 14:08:13.975886', 'class': 'Load()', 'method': 'post()'}
#33 get from queue1 => {'process': 'p1', 'timestamp': '2020-02-24 14:08:16.980382', 'class': 'Load()', 'method': 'post()'}
#62 queue1 empty
</code></pre>
<p><strong>尝试6:使用队列超时</p>
<p>正如我建议的那样,我试图纠正队列超时。该方法同样是队列1从process1到process2,队列2从process2到process1。数据进入队列1,但不返回队列2,它们再次神秘地消失。代码和结果如下所示</p>
<p>尝试6的代码:</p>
<pre><code>#!/usr/bin/python2.7 python2.7
# -*- coding: utf-8 -*-
# script for serialized interprocess data exchange
# common modules
import os
import sys
import time
import uuid
import Queue
#from Queue import Empty
import multiprocessing
from multiprocessing import Process
#from multiprocessing import Queue
from datetime import datetime
someData = {}
class Load():
def post(self):
timestamp = str(datetime.now())
someData = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
queue1.put(someData) # put into queue
print "#24 process 1: put in queue1 =>", someData
time.sleep(3)
# while True: # queue1 checking loop
# if queue1.empty() == False:
# timestamp = str(datetime.now())
# res = queue1.get()
# res = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
# print "#33 get from queue1 =>", res
# break
# else:
# print "#36 queue1 empty"
# time.sleep(1)
while True: # queue2 checking loop
try:
someData = queue2.get(True,1)
timestamp = str(datetime.now())
someData = {"process":"p1","class":"Load()","method":"post()","timestamp":timestamp}
print "#43 process 1: got from queue2 =>", someData
break
except Queue.Empty:
print "#46 process1: queue2 empty"
continue
class Unload():
def get(self):
while True: # queue2 checking loop
try:
someData = queue1.get(True,1)
timestamp = str(datetime.now())
someData = {"process":"p2","class":"Unload()","method":"get()","timestamp":timestamp}
print "#56 process2: got from queue1 =>", someData
break
except Queue.Empty:
print "#59 process2: queue1 empty"
continue
self.doSomething(someData) # call method
def doSomething(self, data):
time.sleep(3)
timestamp = str(datetime.now())
someData = {"process":"p2","class":"Unload()","method":"doSomething()","timestamp":timestamp}
self.someData = someData
print "#68 process2: do something =>", someData
self.put(someData)
def put(self,data):
time.sleep(3)
timestamp = str(datetime.now())
self.someData = {"process":"p2","class":"Unload()","method":"put()","timestamp":timestamp}
someData = self.someData
#print "#81 process 2: put back in queue1 =>", self.someData
#queue1.put(res)
print "#78 process2: put back in queue2 =>", someData
queue2.put(someData)
# main
if __name__ == '__main__':
queue1 = multiprocessing.Queue()
queue2 = multiprocessing.Queue()
global p1, p2
#p1 = Process(target=Load().post(), args=(queue1,)) # process p1
p1 = Process(target=Load().post(), args=(queue1,queue2,))
p1.daemon = True
p1.start()
#p2 = Process(target=Unload().get(), args=(queue1,)) # process p2
p2 = Process(target=Unload().get(), args=(queue1,queue2,))
p2.start()
p2.join()
</code></pre>
<p>尝试6的结果:</p>
<pre><code>#24 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 18:14:46.435661', 'class': 'Load()', 'method': 'post()'}
#46 process1: queue2 empty
</code></pre>
<p>注意:当我在没有类的情况下使用它时,建议的方法是有效的。代码如下:</p>
<pre><code>import uuid
import multiprocessing
from multiprocessing import Process
import Queue
def load(que_in, que_out):
request = {"id": uuid.uuid4(), "workload": "do_stuff", }
que_in.put(request)
print("load: sent request {}: {}".format(request["id"], request["workload"]))
while True:
try:
result = que_out.get(True, 1)
except Queue.Empty:
continue
print("load: got result {}: {}".format(result["id"], result["result"]))
def unload(que_in, que_out):
def processed(request):
return {"id": request["id"], "result": request["workload"] + " processed", }
while True:
try:
request = que_in.get(True, 1)
except Queue.Empty:
continue
print("unload: got request {}: {}".format(request["id"], request["workload"]))
result = processed(request)
que_out.put(result)
print("unload: sent result {}: {}".format(result["id"], result["result"]))
# main
if __name__ == '__main__':
que_in = multiprocessing.Queue()
que_out = multiprocessing.Queue()
p1 = Process(target=load, args=(que_in, que_out)) # process p1
p1.daemon = True
p1.start()
p2 = Process(target=unload, args=(que_in, que_out)) # process p2
p2.start()
p2.join()
</code></pre>
<p><strong>----更新20200225:尝试7-------------------------------------------------------------------------------</strong></p>
<p><strong>尝试7:使用一个队列,在不同类别中出现队列超时(工作)</strong></p>
<p>在这次尝试中,我在不同类的方法之间使用了一个共享队列,并更正了超时。数据从process1传输到proces在共享队列中从进程2返回到进程1。在这次尝试中,数据传输正确。代码和结果如下所示</p>
<p>尝试7的代码:</p>
<pre><code>import uuid
import multiprocessing
from multiprocessing import Process
import Queue
class Input():
def load(self, shared_queue):
request = {"id": uuid.uuid4(), "workload": "do_stuff", }
shared_queue.put(request)
print("load: sent request {}: {}".format(request["id"], request["workload"]))
while True:
try:
result = shared_queue.get(True, 1)
except Queue.Empty:
continue
print("load: got result {}: {}".format(result["id"], result["result"]))
break
class Output():
def unload(self, shared_queue):
def processed(request):
return {"id": request["id"], "result": request["workload"] + " processed", }
while True:
try:
request = shared_queue.get(True, 1)
except Queue.Empty:
continue
print("unload: got request {}: {}".format(request["id"], request["workload"]))
result = processed(request)
shared_queue.put(result)
print("unload: sent result {}: {}".format(result["id"], result["result"]))
# main
if __name__ == '__main__':
shared_queue = multiprocessing.Queue()
up = Input()
down = Output()
p1 = Process(target=up.load, args=(shared_queue,)) # process p1
p1.daemon = True
p1.start()
p2 = Process(target=down.unload, args=(shared_queue,)) # process p2
p2.start()
p1.join()
p2.join()
</code></pre>
<p>尝试7的结果:</p>
<pre><code>load: sent request a461357a-b39a-43c4-89a8-a77486a5bf45: do_stuff
unload: got request a461357a-b39a-43c4-89a8-a77486a5bf45: do_stuff
unload: sent result a461357a-b39a-43c4-89a8-a77486a5bf45: do_stuff processed
load: got result a461357a-b39a-43c4-89a8-a77486a5bf45: do_stuff processed
</code></pre>