многопроцессорная очередь, тупик при порождении нескольких потоков в одном процессе - PullRequest
Новая
       15

многопроцессорная очередь, тупик при порождении нескольких потоков в одном процессе

0 голосов
/ 25 мая 2018

Я создал два процесса, один процесс, который порождает многопоточность, является ответом для записи данных в очередь, другой читает данные из очереди.Это всегда тупик в высокой частоте, меньше нет.Особенно, когда вы добавляете sleep в метод run в модуле записи (комментарий в кодах).Позвольте мне поставить свои коды ниже:

окружения: python2.7

main.py 

    from multiprocessing import Process,Queue
    from write import write
    from read import read

    if __name__ == "__main__":
        record_queue = Queue()
        table_queue = Queue()

        pw = Process(target=write,args=[record_queue, table_queue])
        pr = Process(target=read,args=[record_queue, table_queue])

        pw.start()
        pr.start()
        pw.join()
        pr.join()

write.py 

from concurrent.futures import ThreadPoolExecutor, as_completed

def write(record_queue, table_queue):
    thread_num = 3
    pool = ThreadPoolExecutor(thread_num)
    futures = [pool.submit(run, record_queue, table_queue) for _ in range (thread_num)]
    results = [r.result() for r in as_completed(futures)]


def run(record_queue, table_queue):
    while True:
        if  table_queue.empty():
            break
    table = table_queue.get()
    # adding this code below reduce deadlock opportunity.
    #import time
    #import random
    #time.sleep(random.randint(1, 3))
    process_with_table(record_queue, table_queue, table)

def process_with_table(record_queue, table_queue, table):
    #for short
    for item in [x for x in range(1000)]:
        record_queue.put(item)

read.py 

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
import Queue

def read(record_queue, table_queue):
    count = 0
    while True:

        item = record_queue.get()
        count += 1
        print ("item: ", item)
        if count == 4:
            break

Я его погуглил и на SO такие же вопросы, но я не вижу сходства по сравнению с моим кодом, так что кто-нибудь может помочь с моими кодами, спасибо ...

1 Ответ

0 голосов
/ 25 мая 2018

Кажется, я нашел решение, изменив метод запуска в модуле записи на: 

def run(record_queue, table_queue):
    while True:
        try:  
             if table_queue.empty():
                 break
             table = table_queue.get(timeout=3)

             process_with_table(record_queue, table_queue, table)
        except multiprocessing.queues.Empty:
             import time
             time.sleep(0.1)

и никогда не вижу тупиков или блокировок в методе get.

...