当前位置 主页 > 网站技术 > 代码类 >

    Python 使用threading+Queue实现线程池示例

    栏目:代码类 时间:2019-12-21 18:10

    一、线程池

    1、为什么需要使用线程池

    1.1 创建/销毁线程伴随着系统开销,过于频繁的创建/销毁线程,会很大程度上影响处理效率。

    记创建线程消耗时间T1,执行任务消耗时间T2,销毁线程消耗时间T3,如果T1+T3>T2,那说明开启一个线程来执行这个任务太不划算了!在线程池缓存线程可用已有的闲置线程来执行新任务,避免了创建/销毁带来的系统开销。

    1.2 线程并发数量过多,抢占系统资源从而导致阻塞。

    线程能共享系统资源,如果同时执行的线程过多,就有可能导致系统资源不足而产生阻塞的情况。

    1.3 对线程进行一些简单的管理。

    比如:延时执行、定时循环执行的策略等,运用线程池都能进行很好的实现。

    2、Python中建立线程池的方法

    2.1 使用threadpool模块,这是个python的第三方模块,支持python2和python3

    2.2 使用concurrent.futures模块,这个模块是python3中自带的模块,python2.7以上版本也可以安装使用

    2.3 自己构建一个线程池

    二、队列(queue)

    Queue模块提供的队列(FIFO)适用于多线程编程,在生产者(producer)和消费者(consumer)之间线程安全(thread-safe)地传递消息或其它数据,因此多个线程可以共用同一个Queue实例。常用方法:

    Queue.qsize():返回queue的大小。

    Queue.empty():判断队列是否为空,通常不太靠谱。

    Queue.full():判断是否满了。

    Queue.put(item, block=True, timeout=None): 往队列里放数据。

    Queue.put_nowait(item):往队列里存放元素,不等待

    Queue.get(item, block=True, timeout=None): 从队列里取数据。

    Queue.get_nowait(item):从队列里取元素,不等待

    Queue.task_done():表示队列中某个元素是否的使用情况,使用结束会发送信息。

    Queue.join():一直阻塞直到队列中的所有元素都执行完毕。

    三、使用threading+Queue处理多任务

    假设有十个任务需要处理,打算在后台开启五个线程,简化后的模型

    import Queue
    import threading
    import time
     
    queue = Queue.Queue()
     
    class ThreadNum(threading.Thread):
      def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
     
      def run(self):
        while True:
          #消费者端,从队列中获取num
          num = self.queue.get()
          print("Retrieved", num)
          time.sleep(1) 
          #在完成这项工作之后,使用 queue.task_done() 函数向任务已经完成的队列发送一个信号
          self.queue.task_done()
        
        print("Consumer Finished")
     
    def main():
      #产生一个 threads pool, 并把消息传递给thread函数进行处理,这里开启10个并发
      for i in range(5):
        t = ThreadNum(queue)
        t.setDaemon(True)
        t.start()
      
      #往队列中填数据 
      for num in range(10):
        queue.put(num)
        #wait on the queue until everything has been processed
      
      queue.join()
       
    if __name__ == '__main__':
      main()
      time.sleep(500)

    输出为:

    ('Retrieved', 0)
     ('Retrieved', 1)('Retrieved', 2)
    ('Retrieved', 3)
    ('Retrieved', 4)
    ('Retrieved', 5)('Retrieved', 6)
    ('Retrieved', 7)
    ('Retrieved', 8)
     ('Retrieved', 9)

    具体工作步骤描述如下:

    1、创建一个 Queue.Queue() 的实例,然后使用数据对它进行填充。

    2、将经过填充数据的实例传递给线程类,后者是通过继承 threading.Thread 的方式创建的。