Python之路——线程池

1 线程基础

  1.1 线程状态

  线程有5种状态，状态转换的过程如下图所示：

  1.2 线程同步——锁

      多线程的优势在于可以同时运行多个任务（至少感觉起来是这样，其实Python中是伪多线程）。但是当线程需要共享数据时，可能存在数据不同步的问题。考虑这样一种情况：一个列表里所有元素都是0，线程"set"从后向前把所有元素改成1，而线程"print"负责从前往后读取列表并打印。那么，可能线程"set"开始改的时候，线程"print"便来打印列表了，输出就成了一半0一半1，这就是数据的不同步。为了避免这种情况，引入了锁的概念。

    锁有两种状态—锁定和未锁定。每当一个线程比如"set"要访问共享数据时，必须先获得锁定；如果已经有别的线程比如"print"获得锁定了，那么就让线程"set"暂停，也就是同步阻塞；等到线程"print"访问完毕，释放锁以后，再让线程"set"继续。经过这样的处理，打印列表时要么全部输出0， 要么全部输出1，不会再出现一半0一半1的尴尬场面。

    线程与锁的交互如下图所示：

  1.3 线程通信（条件变量）

    然而还有另外一种尴尬的情况：列表并不是一开始就有的；而是通过线程"create"创建 的。如果"set"或者"print" 在"create"还没有运行的时候就访问列表，将会出现一个异常。使用锁可以解决这个问题，但是"set"和"print"将需要一个无限循环——他们不知道"create"什么时候会运行，让"create"在运行后通知"set"和"print"显然是一个更好的解决方案。于是，引入了条件变量。

    条件变量允许线程比如"set"和"print"在条件不满足的时候（列表为None时）等待，等到条件满足的时候（列表已经创建）发出一个通知，告诉"set" 和"print"条件已经有了，你们该起床干活了；然后"set"和"print"才继续运行。

   线程与条件变量的交互如下图所示：

   1.4 线程运行和阻塞的状态转换

        最后看看线程运行和阻塞状态的转换。

     阻塞有三种情况：
     同步阻塞是指处于竞争锁定的状态，线程请求锁定时将进入这个状态，一旦成功获得锁定又恢复到运行状态；
     等待阻塞是指等待其他线程通知的状态，线程获得条件锁定后，调用“等待”将进入这个状态，一旦其他线程发出通知，线程将进入同步阻塞状态，再次竞争条件锁定；
     而其他阻塞是指调用time.sleep()、anotherthread.join()或等待IO时的阻塞，这个状态下线程不会释放已获得的锁定。

2 多线程

     为充分利用cpu资源，减少资源浪费，Python中也提供了多线程技术，但由于存在GIL的关系，实际上同一时刻只存在一个线程在执行，即Python中不存在真正意义上的多线程，但是多线程技术还是提高了CPU资源的利用率，所以还是很有价值的。

3 线程池

    我们知道系统处理任务时，需要为每个请求创建和销毁对象。当有大量并发任务需要处理时，再使用传统的多线程就会造成大量的资源创建销毁导致服务器效率的下降。这时候，线程池就派上用场了。线程池技术为线程创建、销毁的开销问题和系统资源不足问题提供了很好的解决方案。

   3.1 优点：

   （1）可以控制产生线程的数量。通过预先创建一定数量的工作线程并限制其数量，控制线程对象的内存消耗。

   （2）降低系统开销和资源消耗。通过对多个请求重用线程，线程创建、销毁的开销被分摊到了多个请求上。另外通过限制线程数量，降低虚拟机在垃圾回收方面的开销。

   （3）提高系统响应速度。线程事先已被创建，请求到达时可直接进行处理，消除了因线程创建所带来的延迟，另外多个线程可并发处理。

   “池”的概念使得人们可以定制一定量的资源，然后对这些资源进行复用，而不是频繁的创建和销毁。

   3.2 注意事项

    虽然线程池是构建多线程应用程序的强大机制，但使用它并不是没有风险的。在使用线程池时需注意线程池大小与性能的关系，注意并发风险、死锁、资源不足和线程泄漏等问题。
    1、线程池大小。多线程应用并非线程越多越好，需要根据系统运行的软硬件环境以及应用本身的特点决定线程池的大小。一般来说，如果代码结构合理的话，线程数目与CPU 数量相适合即可。如果线程运行时可能出现阻塞现象，可相应增加池的大小；如有必要可采用自适应算法来动态调整线程池的大小，以提高CPU 的有效利用率和系统的整体性能。
    2、并发错误。多线程应用要特别注意并发错误，要从逻辑上保证程序的正确性，注意避免死锁现象的发生。
    3、线程泄漏。这是线程池应用中一个严重的问题，当任务执行完毕而线程没能返回池中就会发生线程泄漏现象。

  3.3 线程池设计的关键点

   1、取任务数与线程数的最小值来决定开启线程的数量，即min(任务数，线程数)；

    2、当前线程池中线程的状态，即正在运行的线程数量和正在等待的线程数量；

    3、关闭线程。

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import Queue
import threading
import time

'''
要点：
1、利用Queue队列特性，将创建的线程对象放入队列中；
2、当执行任务时，从queue队列中取线程执行任务；
3、当任务执行完毕后，线程放回线程池
'''

#线程池类
class ThreadPool(object): 
    def __init__(self,max_thread=20):  #默认最大线程数为20
        self.queue = Queue.Queue(max_thread) #创建队列，大小为max_thread
        #把线程对象（线程类名）加入到队列中，此时并没有创建线程对象，只占用很小内存
        for i in xrange(max_thread):
            self.queue.put(threading.Thread) #注意此处是类名，如果这里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) #创建大小为10的线程池

#任务函数
def func(arg,p):
    print arg
    time.sleep(2)
    p.add_thread() #当前线程执行完，在队列里增加一个线程

for i in xrange(300):
    #获取队列中的线程对象(此时线程对象还没创建)，默认queue.get()，如果队列里没有线程就会阻塞等待。
    thread = pool.get_thread() 
    t = thread(target=func,args=(i,pool)) #此时才真正创建线程对象，并传递参数
    t.start()  #开启线程

'''
for i in xrange(300):
    thread = pool.get_thread()
    t = thread(target=func,args=(i,pool))
    t.start()
    在threading.Thread的构造函数中：self.__args = args  self.__target = target
    
#当然也可以通过下面方法赋值，但是不推荐(对于私有字段，一般建议在构造函数中传参赋值)
for i in xrange(300):
    ret = pool.get_thread()
    ret._Thread__target = func
    ret._Thread__args = (i,pool)
    ret.start()
'''

    3.4  复杂线程池预备知识

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import Queue

obj = object() #先创建一个object对象obj
q = Queue.Queue()  #创建队列

#把obj对象放入队列，注意此处与上述例子中的区别，上例是每次都创建对象，那么每个对象都会占用内存；
#而此处是先创建对象，再把此对象加入队列，即对象只创建了1次，而队列中加入了10次，那么对象只会占用一份内存
for i in range(10):
    q.put(obj) 
    
for i in range(10):
    print id(q.get())  #实验结果是此处id完全相同

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import contextlib
import threading
import time
import random

doing = [] #线程列表

#打印正在运行的线程个数函数
def num(l2):
    while True:
        print len(l2)
        time.sleep(1)

#单独创建一个线程，用于打印当前正在运行的线程数量
t = threading.Thread(target=num,args=(doing,))  
t.start()

#管理上下文的装饰器
@contextlib.contextmanager
def show(l1,item):
    l1.append(item)  #把正在执行的当前线程加入列表
    yield  #运行到此处，程序就会跳出当前函数，去执行with关键字中的内容，执行完后在进入当前函数执行
    l1.remove(item)  #当前线程执行完成后，则从列表中移除

#任务函数
def task(i):
    flag = threading.current_thread()
    with show(doing,flag):  #with管理上下文，进行切换
        print len(doing)
        time.sleep(random.randint(1,4)) #等待

for i in range(20): #创建20个线程
    temp = threading.Thread(target=task,args=(i,))
    temp.start()

更多with上下文管理知识请参考：

        http://www.cnblogs.com/alan-babyblog/p/5153343.html

        http://www.cnblogs.com/alan-babyblog/p/5153386.html

   3.5 twisted源码中经典线程池的实现

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from Queue import Queue
import contextlib
import threading

WorkerStop = object()

class ThreadPool:
    workers = 0
    threadFactory = threading.Thread  #类名
    currentThread = staticmethod(threading.currentThread) #静态方法

    def __init__(self, maxthreads=20, name=None):
        self.q = Queue(0) #创建队列，参数0表示队列大小不限，队列用于存放任务
        self.max = maxthreads #定义最大线程数
        self.name = name
        self.waiters = [] #存放等待线程的列表
        self.working = [] #存放工作线程的列表

    def start(self):
        needSize = self.q.qsize()  #获取任务所需的线程数
        while self.workers < min(self.max, needSize): #wokers初始值为0
            self.startAWorker()  #调用开启线程的方法

    def startAWorker(self):
        self.workers += 1  #workers自增1
        newThread = self.threadFactory(target=self._worker, name='test') #创建1个线程并去执行_worker方法
        newThread.start()

    def callInThread(self, func, *args, **kw):
        self.callInThreadWithCallback(None, func, *args, **kw)

    def callInThreadWithCallback(self, onResult, func, *args, **kw):
        o = (func, args, kw, onResult)  #把参数组合成元组 
        self.q.put(o)  #把任务加入队列

    #上下文管理装饰器    
    @contextlib.contextmanager
    def _workerState(self, stateList, workerThread):
        stateList.append(workerThread)  #把当前执行线程加入线程状态类表stateList
        try:
            yield
        finally:
            stateList.remove(workerThread)  #执行完后移除

    def _worker(self):
        ct = self.currentThread() #获取当前线程id
        o = self.q.get() #队列中获取执行任务
        
        while o is not WorkerStop:  #当获取的o不是workstop信号时，执行while循环
            with self._workerState(self.working, ct):  #上下文切换
                function, args, kwargs, onResult = o  #元组分别赋值给变量
                del o  #删除元组o
                try:
                    result = function(*args, **kwargs)  
                    success = True
                except:
                    success = False
                    if onResult is None:
                        pass
                    else:
                        pass

                del function, args, kwargs

                if onResult is not None:
                    try:
                        onResult(success, result)
                    except:
                        #context.call(ctx, log.err)
                        pass

                del onResult, result

            with self._workerState(self.waiters, ct): 
                o = self.q.get()  #获取任务

    def stop(self): #关闭线程
        while self.workers: #循环workers
            self.q.put(WorkerStop) #在队列中增加一个信号~
            self.workers -= 1 #workers值-1，直到所有线程关闭

def show(arg):
    import time
    time.sleep(1)
    print arg

pool = ThreadPool(10)

#创建500个任务，队列里添加500个任务
#每个任务都是一个元组（方法名，动态参数，动态参数，默认为NoNe）
for i in range(100):
    pool.callInThread(show, i)

pool.start() #开启执行任务

pool.stop()  #执行关闭线程方法

 

参考资料：

       http://www.cnblogs.com/huxi/archive/2010/06/26/1765808.html

     http://www.cnblogs.com/wupeiqi/articles/4839959.html