zoukankan      html  css  js  c++  java
  • python2 里边自定义线程池

    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import Queue
    import threading
    
    
    class ThreadPool(object):
    
        def __init__(self, max_num=20):
            self.queue = Queue.Queue(max_num)
            for i in xrange(max_num):
                self.queue.put(threading.Thread)
    
        def get_thread(self):
            return self.queue.get()
    
        def add_thread(self):
            self.queue.put(threading.Thread)
    
    """
    pool = ThreadPool(10)
    
    def func(arg, p):
        print arg
        import time
        time.sleep(2)
        p.add_thread()
    
    
    for i in xrange(30):
        thread = pool.get_thread()
        t = thread(target=func, args=(i, pool))
        t.start()
    """

    版本2:

    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    
    import queue
    import threading
    import contextlib
    import time
    
    StopEvent = object()
    
    
    class ThreadPool(object):
    
        def __init__(self, max_num, max_task_num = None):
            if max_task_num:
                self.q = queue.Queue(max_task_num)
            else:
                self.q = queue.Queue()
            self.max_num = max_num
            self.cancel = False
            self.terminal = False
            self.generate_list = []
            self.free_list = []
    
        def run(self, func, args, callback=None):
            """
            线程池执行一个任务
            :param func: 任务函数
            :param args: 任务函数所需参数
            :param callback: 任务执行失败或成功后执行的回调函数,回调函数有两个参数1、任务函数执行状态;2、任务函数返回值(默认为None,即:不执行回调函数)
            :return: 如果线程池已经终止,则返回True否则None
            """
            if self.cancel:
                return
            if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
                self.generate_thread()
            w = (func, args, callback,)
            self.q.put(w)
    
        def generate_thread(self):
            """
            创建一个线程
            """
            t = threading.Thread(target=self.call)
            t.start()
    
        def call(self):
            """
            循环去获取任务函数并执行任务函数
            """
            current_thread = threading.currentThread()
            self.generate_list.append(current_thread)
    
            event = self.q.get()
            while event != StopEvent:
    
                func, arguments, callback = event
                try:
                    result = func(*arguments)
                    success = True
                except Exception as e:
                    success = False
                    result = None
    
                if callback is not None:
                    try:
                        callback(success, result)
                    except Exception as e:
                        pass
    
                with self.worker_state(self.free_list, current_thread):
                    if self.terminal:
                        event = StopEvent
                    else:
                        event = self.q.get()
            else:
    
                self.generate_list.remove(current_thread)
    
        def close(self):
            """
            执行完所有的任务后,所有线程停止
            """
            self.cancel = True
            full_size = len(self.generate_list)
            while full_size:
                self.q.put(StopEvent)
                full_size -= 1
    
        def terminate(self):
            """
            无论是否还有任务,终止线程
            """
            self.terminal = True
    
            while self.generate_list:
                self.q.put(StopEvent)
    
            self.q.queue.clear()
    
        @contextlib.contextmanager
        def worker_state(self, state_list, worker_thread):
            """
            用于记录线程中正在等待的线程数
            """
            state_list.append(worker_thread)
            try:
                yield
            finally:
                state_list.remove(worker_thread)
    
    
    
    # How to use
    
    
    pool = ThreadPool(5)
    
    def callback(status, result):
        # status, execute action status
        # result, execute action return value
        pass
    
    
    def action(i):
        print(i)
    
    for i in range(30):
        ret = pool.run(action, (i,), callback)
    
    time.sleep(5)
    print(len(pool.generate_list), len(pool.free_list))
    print(len(pool.generate_list), len(pool.free_list))
    # pool.close()
    # pool.terminate()
    View Code

    更多参见:twisted.python.threadpool

    上下文管理:https://docs.python.org/2/library/contextlib.html

    对于IO请求,用多线程性能好点;而对于Python全局解释器锁GIL

    它本身只是针对与CPU调度的时候,才会有严格的限制;而对于 IO的请求

    GIL的影响不是很大。

    对于计算密集型的话,由于GIL的限制,一个CPU只能服务一个线程,

    会有阻塞,所以适合用多进程;来实现多个进程共用一个CPU。

    而对于上述,最优的方案应该是协程,

    使用单个进程中的一个线程完成多个任务,实现并发,就是协程!

    注意:协程中,还有微线程。

    例如:话务员给客户打电话。

    GIL全局解释器锁: 限制同一时刻,CPU只能调度进程中的一个线程!

    grequests实现

    select.select 

    select的原型为(rlist,wlist,xlist[,timeout]),其中rlist是等待读取的对象,wlist是等待写入的对象,xlist是等待异常的对象,最后一个是可选对象,指定等待的时间

  • 相关阅读:
    2015.2.27 UltraEdit中显示XML结构
    2015.1.31 DataGridView自动滚动到某行
    2015.1.15 利用函数实现将一行记录拆分成多行记录 (多年想要的效果)
    2015.1.15 利用Oracle函数返回表结果 重大技术进步!
    2015.1.15 利用Oracle函数插入表结构 Bulk collect into 不用循环,简洁高效
    2015.1.8 Left join 左连接
    2015.1.10 解决DataGridView SelectionChanged事件自动触发问题
    delphi 遍历窗口
    delphi 访问 protected 属性 哈哈
    clientdataset 读取excel 如果excel 文件不存在的时候 相应的gird 会不显示数据, 鼠标掠过 gird 格子 才会显示数据。 这是一个bug 哈哈
  • 原文地址:https://www.cnblogs.com/hellojesson/p/6400415.html
Copyright © 2011-2022 走看看