Как я могу настроить многоканальный конвейер Python (моя попытка использует очередь и поток) - PullRequest
Новая
       17

Как я могу настроить многоканальный конвейер Python (моя попытка использует очередь и поток)

0 голосов
/ 27 апреля 2019

Как я могу настроить многоканальный конвейер в python, где после заполнения до определенной точки программа будет извлекать данные со всех каналов в режиме FIFO.Я думаю, что решение должно использовать очередь и многопоточность.

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

import numpy as np
from queue import Queue
from threading import Thread
import time



class qtest():
    def __init__(self):
        self.q = Queue()
        self.num_threads = 10 
        threads = []
        for i in range(self.num_threads):
          worker = Thread(target=self.pop)
          worker.setDaemon(True)
          worker.start()
          threads.append(worker)
        self.threads = threads   

    def pop(self):
      #fill the threads. 
      #assuming that each thread will be filled at a random speed
      #I want to pop all in a FIFO manner when all are filled to a certain point.
        while True:
          item = self.q.get()
          if item is None:
            break
          self.q.task_done()


    def randWait(self, entry):
        #Do the work.  Fill the pipe (thread)
        time.sleep(np.random.randint(5))
        self.q.put(entry)


    def start(self):
        for t in range(self.num_threads):
            print("Starting thread %s"%(t))
            self.randWait(t)
        self.q.join()


    def stop(self):
        # stop workers
        for i in range(self.num_threads):
            self.q.put(None)
        for t in self.threads:
            t.join()
        print("closed!")


def main():
    h = qtest()
    h.start()
    h.stop()

if __name__ == '__main__':
  main()
...