Как правильно использовать очереди в python?

Question

Я новичок в том, что касается python многопоточности и многопроцессорности, поэтому, пожалуйста, потерпите меня.

Я хочу создать систему, состоящую из трех python сценариев. Первый создает некоторые данные и непрерывно отправляет эти данные во второй скрипт. Второй сценарий берет данные и сохраняет в некоторый файл, пока файл не превысит определенный лимит памяти. Когда это происходит, третий скрипт отправляет данные на внешнее устройство и избавляется от этого «кэша». Мне нужно, чтобы все это происходило одновременно. Псевдокод подводит итог того, что я пытаюсь сделать.

def main_1():

  data = [1,2,3]
  send_to_second_script(data)

def main_2():

  rec_data = receive_from_first_script()
  save_to_file(rec_data)
  if file>limit:
     signal_third_script()

def main_3():
  if signal is true:
     send_data_to_external_device()
     remove_data_from_disk()

Я понимаю, что могу использовать очереди, чтобы это произошло, но я не уверен, как.

Кроме того, пока что Сделав это, я попробовал другой подход, где я создал один скрипт python и использовал потоки для порождения потоков для каждой части процесса. Это правильно или использование очередей лучше?

Answer 1

Во-первых, для Python вы должны по-настоящему осознавать, что дает вам преимущества многопоточности / многопроцессорности. ИМО, вы должны рассмотреть многопроцессорность вместо многопоточности. Потоки в Python на самом деле не параллельны из-за GIL, и есть много объяснений, какой из них использовать. Самый простой способ выбрать - посмотреть, привязана ли ваша программа к IO или CPU. В любом случае, в очередь, которая является простым способом работы с несколькими процессами в python.

Используя ваш псевдокод в качестве примера, вот как вы должны использовать Очередь.

import multiprocessing



def main_1(output_queue):
    test = 0
    while test <=10: # simple limit to not run forever
        data = [1,2,3]
        print("Process 1: Sending data")
        output_queue.put(data) #Puts data in queue FIFO
        test+=1
    output_queue.put("EXIT") # triggers the exit clause

def main_2(input_queue,output_queue):
    file = 0 # Dummy psuedo variables
    limit = 1
    while True:
        rec_data = input_queue.get() # Get the latest data from queue. Blocking if empty
        if rec_data == "EXIT": # Exit clause is a way to cleanly shut down your processes
            output_queue.put("EXIT")
            print("Process 2: exiting")
            break
        print("Process 2: saving to file:", rec_data, "count = ", file)
        file += 1
        #save_to_file(rec_data)
        if file>limit:
            file = 0 
            output_queue.put(True)

def main_3(input_queue):
    while(True):
        signal = input_queue.get()

        if signal is True:
            print("Process 3: Data sent and removed")
            #send_data_to_external_device()
            #remove_data_from_disk()
        elif signal == "EXIT":
            print("Process 3: Exiting")
            break

if __name__== '__main__':

    q1 = multiprocessing.Queue() # Intializing the queues and the processes
    q2 = multiprocessing.Queue()
    p1 = multiprocessing.Process(target = main_1,args = (q1,))
    p2 = multiprocessing.Process(target = main_2,args = (q1,q2,))
    p3 = multiprocessing.Process(target = main_3,args = (q2,))

    p = [p1,p2,p3]
    for i in p: # Start all processes
        i.start()
    for i in p: # Ensure all processes are finished
        i.join()

Отпечатки могут быть немного не в порядке, потому что я не удосужился заблокировать std_out. Но использование очереди гарантирует, что материал перемещается из одного процесса в другой.

РЕДАКТИРОВАТЬ: НЕОБХОДИМО помнить, что вам также следует обратить внимание на многопроцессорные блокировки, чтобы гарантировать, что ваш файл является «поточно-ориентированным» при выполнении перемещения. /удалять. Приведенный выше псевдокод демонстрирует, как использовать очередь