Как обмениваться данными между двумя отдельными процессами в Python - PullRequest
Новая
       117

Как обмениваться данными между двумя отдельными процессами в Python

2 голосов
/ 23 февраля 2020

ПРОБЛЕМА

Параллельно выполняются два отдельных процесса, и я хотел бы, чтобы они общались взад и вперед.

ОБЪЯСНЕНИЕ КОД

Код в Python2 .7. В моем сценарии с минимальным разрешением я использую очередь для связи между процессами. Процесс p1 помещает данные в очередь. Процесс p2 получает данные из очереди и что-то делает с данными. Затем процесс p2 помещает измененные данные обратно в очередь и, наконец, затем процесс p1 возвращает измененные данные из очереди. Измененные данные должны возвращаться процессу p1, поскольку этот процесс действительно является сервером событий, который отправляет / получает запросы.

КОД 

#!/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()

ВОПРОС Я проверил другие ресурсы в отношении, но все они связаны с общением в одном направлении. Ниже приведен список ресурсов.

  1. use-get-nowait-in- python -without -ising-empty-exception
  2. in- python -про-как-то-вы-получите-возвращаем-данные-от-конкретного-процесса-использования-многопроцессора
  3. как использовать-многопроцессорную очередь -с блокировкой
  4. многопроцессорный модуль поддерживает блокировки
  5. нить, которую я могу приостановить и возобновить
  6. обмен-данными-между-двумя- python -процессами

Как мне заставить процесс1 ждать и получать измененные данные из процесса2? Должен ли я рассмотреть другой подход для связи между процессами, например, pipe, zeroMQ?

ПОПЫТКА 1: использование time.sleep () без циклов while в процессе 1 только с time.sleep данные go находятся в очереди, но никогда не достигают конечного пункта назначения в процессе 1. Пока все хорошо, но последний шаг отсутствует. Результаты приведены ниже. 

#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()'}

ПОПЫТКА 2: использование while l oop в процессе 1 С while l oop проверка очереди для данных go в очереди, но попадают сразу после этого, они никогда не достигают процесса 2. Результаты приведены ниже. 

#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

ПОПЫТКА 3: использование двух очередей использование двух очередей: queue1 из process1 в process2, очередь2 из процесса2 в процесс1. Данные go в очереди1, но не возвращаются в очередь2, они таинственно исчезают sh. Результаты приведены ниже. 

#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-23 11:53:39.745177', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty

----- ОБНОВЛЕНИЕ 20200224: попытки 4, 5 и 6 ------------------- ----------------------------------------------

ПОПЫТКА 4: использование двух очередей с менеджером. Queue ()

использование двух очередей с менеджером. Queue (): очередь 1 из процесса 1 в процесс 2, очередь 2 из процесса 2 в process1. Данные go в очереди1, но не возвращаются в очередь2, опять же они загадочно исчезают sh. Код и результаты приведены ниже.

Код попытки 4: #! / Usr / bin / python2 .7 python2 .7 # - - кодирование: utf-8 - - # скрипт для сериализованного межпроцессного обмена данными 

# 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()

Результаты попытки 4: 

#20 process 1: put in queue1 => {'process': 'p1', 'timestamp': '2020-02-24 13:06:17.687762', 'class': 'Load()', 'method': 'post()'}
#42 queue2 empty

ПОПЫТКА 5: использование одной очереди с manager.Queue () Использование одной очереди с manager.Queue (): очередь1 из процесса1 в процесс2, очередь1 обратно из процесса2 в процесс1. Данные go в очереди 1, но сразу после этого перехватываются, они никогда не доходят до процесса 2. Ниже приведены результаты кода.

Код попытки 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()

Результат попытки 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

ПОПЫТКА 6: использование тайм-аутов очереди

Как и предлагалось, я пытался исправить тайм-ауты очереди. Подход - снова очередь1 от процесса1 к процессу2, очередь2 от процесса2 к процессу1. Данные go в очереди1, но не возвращаются в очередь2, опять же они таинственно исчезают sh. Код и результаты приведены ниже.

Код попытки 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()

Результаты попытки 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

ПРИМЕЧАНИЕ. Предлагаемый подход работает, когда я использую это без классов. Код ниже: 

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()

----- ОБНОВЛЕНИЕ 20200225: попытка 7 ---------------------------------- --------------------------------------------

ПОПЫТКА 7: использование одной очереди с таймаутами очереди в разных классах (работает)

В этой попытке я использую одну общую очередь между методами разных классов с исправленными таймаутами. Данные передаются из процесса1 в процесс2 и обратно из процесса2 в процесс1 в shared_queue. В этой попытке данные перемещались правильно. Код и результаты приведены ниже.

Код попытки 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()

Результаты попытки 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

Ответы [ 3 ]

1 голос
/ 23 февраля 2020

Я думаю, вы просто пропустили использование таймаутов очереди 

try:
    result = que_out.get(True, 1)
except queue.Empty:
    continue

Этот упрощенный пример может вам помочь: 

import uuid
from multiprocessing import Process
from multiprocessing import Queue
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 = Queue()
    que_out = 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()

Вывод 

load: sent request d9894e41-3e8a-4474-9563-1a99797bc722: do_stuff
unload: got request d9894e41-3e8a-4474-9563-1a99797bc722: do_stuff
unload: sent result d9894e41-3e8a-4474-9563-1a99797bc722: do_stuff processed
load: got result d9894e41-3e8a-4474-9563-1a99797bc722: do_stuff processed
0 голосов
/ 26 февраля 2020

РЕШЕНИЕ: использование одной общей очереди

Я решил проблему после того, как следовал советам и внес некоторые корректировки, чтобы получить правильный таргетинг для различных методов классов. Обратный поток данных между двумя отдельными процессами теперь корректен. Важным примечанием для меня является уделение дополнительного внимания пакету someData, которым обмениваются два отдельных процесса, это действительно должен быть тот же пакет, который отбрасывается. Отсюда ввод идентификатора "id": uuid.uuid4(), чтобы проверить, совпадает ли пакет с каждым пассажем.

0 голосов
/ 23 февраля 2020

Для распространения изменений между процессами необходимо использовать очередь (и), обернутые менеджером, в противном случае каждый процесс имеет свой отдельный объект очереди и не может видеть другой. Диспетчер создает общий экземпляр очереди для всех дочерних процессов.

Таким образом, queue1 = Queue() становится queue1 = manager.Queue() с from multiprocessing import Manager наверху. Если вы хотите использовать подход с двумя очередями, вам, очевидно, придется обернуть вторую очередь одинаковым образом.

Соответствующие ресурсы:

Несколько очередей из одного диспетчера многопроцессорной обработки

Python документация

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...