Очередь в многопроцессорной обработке заблокирована многопоточностью. Время - PullRequest
Новая
       14

Очередь в многопроцессорной обработке заблокирована многопоточностью. Время

0 голосов
/ 14 февраля 2019

Мне нужно перенести данные из подпроцесса в основной.Подпроцесс при выполнении повторяющейся задачи с использованием threading.timer Когда вызывается threading.timer, queue больше не работает.

Подпроцесс получает данные, а я хочу отображать их в режиме реального времени восновной процесс.

Я написал этот фрагмент, чтобы продемонстрировать проблему: 

import threading
import multiprocessing
class MyClass():
    def __init__(self, q):
        self.q = q
        print("put value in q: ", "start")
        self.q.put("start")
        self.i = 0
        self.update()

    def update(self):
        if self.i < 3:
            print("put value in q: ", self.i)
            self.q.put(self.i)
            self.i += 1
            threading.Timer(0.5, self.update).start()
        else:
            self.stop()

    def stop(self):
        print("put value in q: ", "stop")
        self.q.put("stop")

if __name__ == "__main__":
    q = multiprocessing.Queue()
    process = multiprocessing.Process(target = MyClass, args=(q,))
    process.start()
    process.join()
    for i in range(5):
        print("get value in q: ",q.get(block = True, timeout = 2))

и получаю только это: 

put value in q:  start
put value in q:  0
put value in q:  1
put value in q:  2
put value in q:  stop
get value in q:  start
get value in q:  0

Есть ли решение или обходной путь

1 Ответ

0 голосов
/ 14 февраля 2019

У вас есть process.Он имеет основной поток (MyClass() вызов).threading.Timer() порождает другой поток вместе с основным потоком, поэтому вам придется подождать, пока все дополнительные потоки не прекратятся, прежде чем вы остановите process.Таким образом, чтобы решить проблему, замените threading.Timer(0.5, self.update).start() на (ожидание потоков): 

t = threading.Timer(0.5, self.update)
t.start()
t.join()

или замените threading.Timer(0.5, self.update).start() на (без дополнительных потоков): 

time.sleep(.5)
self.update()

Оба решения должны работать.

...