python - 以不同的速率完成线程并与结果交互

Question

我想获得一些关于从 concurrent.futures.ThreadPoolExecutor 中的多个线程接收数据并迭代结果的方法的反馈。在这种情况下，ThreadPoolExecutor 将 future 的线程结果附加到缓冲容器中，并从同一缓冲容器中读取和退出辅助/解耦操作。

线程管理器工作流程

                    /|-> Thread 1 > results \
ThreadPoolExecutor --|-> Thread 2 > results --> Queue [1,2,3] (end) 
                    \|-> Thread 3 > results /
             

现在我们从先进先出队列容器中的线程获得结果 - 这需要是线程安全的。现在上述过程已完成，结果 (str|int|bool|list|dict|any) 位于容器中，等待下一步处理:传达收集的结果。

沟通工作流程

                                           /|-> Terminal Print
Queue [1,2,3] < Listener > Communicate --|-> Speech Engine Say 
                                           \|-> Write to Log / File

Communicate 类需要在 Queue 上“监听”新条目，并在它们以自己的速度(使用语速一个文本到语音模块 - Producer-Consumer Problem ) 和可能的任何数量的其他输出，所以这真的不能从自上而下调用。如果线程管理器直接调用或让每个线程直接调用 Communicate 类来调用语音引擎，我们将听到断断续续的语音，因为语音引擎将在每次调用时覆盖自身。因此，我们需要将 Thread Manager 工作流与 Communicate 工作流分离，但让它们使用 In/Out 类型缓冲区或 Queue 进行写入和读取，并且需要一个“监听器”概念。

我找到了像下面这样作为守护线程运行的结构的引用，但是 while 循环让我畏缩并且消耗了太多 cpu，所以我仍然需要一种非阻塞方法，其中 self.pipeline 是一个 queue.Queue 对象:

    while True :
        try :
            if not self.pipeline.empty ( ) :    
                task = self.pipeline.get ( timeout=1 )
                if task :       
                    self.serve ( task, )
        except queue.Empty :
            continue

同样，除了 while 循环之外还需要其他东西...

Answer 1

正如您在评论中所写，它是标准的生产者消费者问题。 python 中的一种解决方案是使用多线程和 Queue类(class)队列是线程安全的。它使用 mutex在内部处理忙等待。

Queue.get 最终会调用 wait在其内部互斥锁上。这将阻塞调用线程。但是线程将进入休眠状态，而不是使用 cpu 的忙等待。操作系统的线程调度程序将从此处接管，并在项目可用时唤醒线程(简化)。

因此您仍然可以在多个线程使用者中使用 while True 循环，这些线程使用者在共享队列上调用 queue.get。如果项目可用，线程直接处理它们，如果没有，它们进入休眠模式并释放 cpu。生产者线程也是如此，他们只是调用 Queue.put

但是在 python 中有一个警告。 Python 有一个叫做全局解释器锁的东西——GIL。这是因为它使用了大量的 c 扩展并允许引入 c 扩展的模块。但是那些并不总是线程安全的。 GIL 意味着，一次只能在一个 CPU 上运行一个线程。

因此，一旦一个项目进入队列，一次只有一个消费者会醒来并处理结果。通常也可以一次运行一个生产者。除了那些线程开始等待一些 I/O，比如从套接字读取。因为 I/O 通知是由其他一些 cpu 部分处理的，所以 I/O 总是有一些等待时间。那时，线程释放 GIL，其他线程可以完成工作。

总而言之，如果多个消费者和生产者线程也做一些 I/O 工作——在网络套接字或磁盘上读/写，那么拥有多个消费者和生产者线程才有意义。这称为并发。如果你想同时使用多个cpu核心，你需要在python中使用多处理而不是线程。如果还有一些 IO 工作，那么拥有比内核更多的进程才有意义。

Example