PROBLEMA
Hay dos procesos separados que se ejecutan en paralelo y me gustaría que se comunicaran entre sí.
EXPLICACIÓN DEL CÓDIGO
El código está en Python2.7. En mi script reducido al mínimo, utilizo una cola para la comunicación entre procesos. El proceso p1 pone los datos en una cola. El proceso p2 obtiene los datos de la cola y hace algo con los datos. Luego, el proceso p2 vuelve a colocar los datos modificados en la cola y, finalmente, el proceso p1 recupera los datos modificados de la cola. Los datos modificados deben regresar al proceso p1 porque este proceso realmente es un servidor de eventlet que envía / recibe solicitudes.
CÓDIGO
#!/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()
PREGUNTA He verificado otros recursos al respecto, pero todos involucran comunicación en una dirección. A continuación se muestra la lista de recursos.
¿Cómo consigo que el proceso1 espere y recupere los datos modificados del proceso2? ¿Debería considerar otro enfoque para la comunicación entre procesos, por ejemplo, tuberías, zeroMQ?
INTENTO 1: usar time.sleep () sin los bucles while en el proceso 1 Con solo el time.sleep, los datos suben y retroceden en la cola pero nunca llegan al destino final en el proceso 1. Hasta ahora todo bien, pero el paso final es desaparecido. Los resultados están a continuación.
#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()'}
INTENTO 2: usar el ciclo while en el proceso 1 Con el ciclo while comprobando la cola, los datos van a la cola pero quedan atrapados inmediatamente después, nunca llegan al proceso 2. Los resultados se muestran a continuación.
#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
INTENTO 3: usar dos colas Usar dos colas: cola1 de proceso1 a proceso2, cola2 de proceso2 a proceso1. Los datos van a la cola1 pero no regresan a la cola2, desaparecen misteriosamente. Los resultados están a continuación.
#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:53:39.745177', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty
----- ACTUALIZACIÓN 20200224: intentos 4, 5 y 6 ------------------------------------ -----------------------------
INTENTO 4: usar dos colas con manager.Queue ()
Usando dos colas con el administrador. Cola (): cola1 de proceso1 a proceso2, cola2 de proceso2 a proceso1. Los datos van a la cola1 pero no regresan a la cola2, nuevamente desaparecen misteriosamente. El código y los resultados se encuentran a continuación.
El código del intento 4: #! / Usr / bin / python2.7 python2.7 # - - codificación: utf-8 - - # script para el intercambio de datos entre procesos serializados
# 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()
Los resultados del intento 4:
#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 13:06:17.687762', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty
INTENTO 5: usar una cola con el administrador. Cola () Usar una cola con el administrador. Cola (): cola1 del proceso1 al proceso2, cola1 de regreso del proceso2 al proceso1. Los datos van a la cola1 pero quedan atrapados inmediatamente después, nunca llegan al proceso 2. Los resultados del código se muestran a continuación.
El código del intento 5:
#!/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()
El resultado del intento 5:
#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
INTENTO 6: usar los tiempos de espera de la cola
Como se sugirió, intenté corregir los tiempos de espera de la cola. El enfoque es nuevamente queue1 de process1 a process2, queue2 de process2 a process1. Los datos van a la cola1 pero no regresan a la cola2, nuevamente desaparecen misteriosamente. El código y los resultados se encuentran a continuación.
El código del intento 6:
#!/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()
Los resultados del intento 6:
#24 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 18:14:46.435661', 'class': 'Load()', 'method': 'post()'}
#46 process1: queue2 empty
NOTA: El enfoque sugerido funciona cuando lo uso sin las clases. El código está abajo:
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()
----- ACTUALIZACIÓN 20200225: intento 7 ---------------------------------------- --------------------------------------
INTENTO 7: usar una cola con tiempos de espera de cola en diferentes clases (en funcionamiento)
En este intento, utilizo una cola compartida entre métodos de diferentes clases con los tiempos de espera corregidos. Los datos van de process1 a process2 y regresan de process2 a process1 en una shared_queue. En este intento los datos viajaron correctamente. El código y los resultados se encuentran a continuación.
El código del intento 7:
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()
Los resultados del intento 7:
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
SOLUCIÓN: usar una cola compartida
Resolví el problema después de seguir las sugerencias y hacer algunos ajustes para obtener la orientación correcta de los diferentes métodos de clases. El flujo de ida y vuelta de los datos entre dos procesos separados ahora es correcto. Una nota importante para mí es prestar especial atención al someData
paquete intercambiado entre dos procesos separados, realmente tiene que ser el mismo paquete que se tira. De ahí la entrada del identificador "id": uuid.uuid4()
para comprobar si el paquete es el mismo con cada pasaje.
#!/usr/bin/python2.7 python2.7
# -*- coding: utf-8 -*-
# script for back and forth communication between two separate processes using a shared queue
# common modules
import os
import sys
import time
import uuid
import Queue
import multiprocessing
from multiprocessing import Process
from datetime import datetime
someData = {}
class Load():
def post(self, sharedQueue):
timestamp = str(datetime.now()) # for timing checking
someData = {"timestamp":timestamp, "id": uuid.uuid4(), "workload": "do_stuff",}
self.someData = someData
sharedQueue.put(someData) # put into the shared queue
print("#25 p1 load: sent someData {}: {}".format(someData["id"], someData["timestamp"], someData["workload"]))
time.sleep(1) # for the time flow
while True: # sharedQueue checking loop
try:
time.sleep(1) # for the time flow
timestamp = str(datetime.now())
someData = sharedQueue.get(True,1)
someData["timestamp"] = timestamp
print("#37 p1 load: got back someData {}: {}".format(someData["id"], someData["timestamp"], someData["workload"]))
break
except Queue.Empty:
print("#37 p1: sharedQueue empty")
continue
break
class Unload():
def get(self, sharedQueue):
while True: # sharedQueue checking loop
try:
someData = sharedQueue.get(True,1)
self.someData = someData
timestamp = str(datetime.now())
someData["timestamp"] = timestamp
print("#50 p2 unload: got someData {}: {}".format(someData["id"], someData["timestamp"], someData["workload"]))
break
except Queue.Empty:
print("#53 p2: sharedQueue empty")
continue
time.sleep(1) # for the time flow
self.doSomething(someData) # pass the data to the method
def doSomething(self, someData): # execute some code here
timestamp = str(datetime.now())
someData["timestamp"] = timestamp
print("#62 p2 unload: doSomething {}: {}".format(someData["id"], someData["timestamp"], someData["workload"]))
self.put(someData)
time.sleep(1) # for the time flow
def put(self,someData):
timestamp = str(datetime.now())
someData["timestamp"] = timestamp
sharedQueue.put(someData)
print("#71 p2 unload: put someData {}: {}".format(someData["id"], someData["timestamp"], someData["workload"]))
time.sleep(1) # for the time flow
# main
if __name__ == '__main__':
sharedQueue = multiprocessing.Queue()
trx = Load()
rcx = Unload()
p1 = Process(target=trx.post, args=(sharedQueue,)) # process p1
p1.daemon = True
p1.start()
p2 = Process(target=rcx.get, args=(sharedQueue,)) # process p2
p2.start()
p1.join()
p2.join()
Este artículo se recopila de Internet, indique la fuente cuando se vuelva a imprimir.
En caso de infracción, por favor [email protected] Eliminar
Déjame decir algunas palabras