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是等待异常的对象,最后一个是可选对象,指定等待的时间

  • 相关阅读:
    web print
    datediff
    Typical AJAX
    sql中构建sql语句,再exec这条SQL语句
    ASP.NET中将数据输出到WORD、EXCEL、TXT、HTM
    sql中添加单引号
    sql语句批处理
    日期分组
    异步线程
    电脑内存大有什么好处?
  • 原文地址:https://www.cnblogs.com/hellojesson/p/6400415.html
Copyright © 2011-2022 走看看