python核心编程笔记----threading

一个进程中的各个线程之间共享同一片数据空间,所以线程之间可以比进程之间更方便地共享数据以及相互通讯。

1.全局解释器锁(GIL)

Python 解释器中可以“运行”多个线程,但在任意时刻,只有一个线程在解释器中运行.
在多线程环境中,Python 虚拟机（解释器）按以下方式执行：
1）.设置 GIL
2）.切换到一个线程去运行
3）.运行:
a. 指定数量的字节码指令,或者
b. 线程主动让出控制(可以调用 time.sleep(0))
4）.把线程设置为睡眠状态
5）.解锁 GIL
6）.再次重复以上所有步骤

2.退出线程

当一个线程结束计算,它就退出了。线程可以调用 thread.exit()之类的退出函数,也可以使用Python 退出进程的标准方法,如 sys.exit()或抛出一个 SystemExit 异常等。不过,你不可以直接“杀掉”("kill")一个线程。

3.Python 的线程模块

thread 和threading 模块允许程序员创建和管理线程。
thread 模块提供了基本的线程和锁的支持,而 threading提供了更高级别,功能更强的线程管理的功能。
Queue 模块允许用户创建一个可以用于多个线程之间共享数据的队列数据结构。

4.threading 模块

threading 模块对象 　　　　　　　　　　　　 描述
Thread 　　　　　　　　　　　　 　       表示一个线程的执行的对象
Lock 　　　　　　　　　　　　　　　　 锁原语对象(跟 thread 模块里的锁对象相同)
RLock 　　　　　　　　　　　　　　　  可重入锁对象。使单线程可以再次获得已经获得了的锁(递归锁定)。
Condition 　　　　　　　　　　　　　    条件变量对象能让一个线程停下来,等待其它线程满足了某个“条件”。如,状态的改变或值的改变。
Event 　　　　　　　　　　　　　　       通用的条件变量。多个线程可以等待某个事件的发生,在事件发生后,所有的线程都会被激活。
Semaphore 　　　　　　　　　　　　    为等待锁的线程提供一个类似“等候室”的结构
BoundedSemaphore 　　　　　　　　    与 Semaphore 类似,只是它不允许超过初始值
Timer　　　　　　                                    与Thread 相似,只是,它要等待一段时间后才开始运行

threading中的 Thread类是主要的运行对象

函数
描述
start()                                                       开始线程的执行
run()                                                         定义线程的功能的函数(一般会被子类重写)
join(timeout=None)                                  程序挂起,直到线程结束;如果给了 timeout,则最多阻塞 timeout 秒
getName()                                               返回线程的名字
setName(name)                                      设置线程的名字
isAlive()                                                   布尔标志,表示这个线程是否还在运行中
isDaemon()                                             返回线程的 daemon 标志（守护线程标志）
setDaemon(daemonic)                           把线程的 daemon 标志设为 daemonic(一定要在调用 start()函数前调用)

import threading
from time import sleep, ctime

loopn=[2,4]


def loop(nloop,nsec):
    print "loop "+str(nloop)+" start at: "+ctime()
    sleep(nsec)
    print "loop "+str(nloop)+" stop at: "+ctime()

    
def main():
    print "start at "+ctime()
    thread=[]
    for i in range(len(loopn)):
        t=threading.Thread(target=loop,args=(i,loopn[i]))
        thread.append(t)
        #start，join 在这里执行时执行的线程运行直到结束后才开始第二个线程。
    for i in range(len(loopn)):
        thread[i].start()
        
    for i in range(len(loopn)):
        thread[i].join()   #如果你的主线程除了等线程结束外,还有其它的事情要做(如处理或等待其它的客户请求),那就不用调用 join(),只有在你要等待线程结束的时候才要调用 join()

    print "stop at "+ctime()
    
    
    
if __name__ == '__main__':
    main()

eg:所有的线程都创建了之后,再一起调用 start()函数启动,而不是创建一个启动一个。

join()的作用:

一旦线程启动后,就会一直运行,直到线程的函数结束,退出为止。防止在主线程结束时 子线程还没执行完就结束

5.生产者-消费者问题和 Queue 模块

函数 　　　　　　　　　　　　　　　　　　　　 描述
Queue 模块函数
queue(size)                                            创建一个大小为 size 的 Queue 对象
Queue                                                    对象函数
qsize()                                                    返回队列的大小(由于在返回的时候,队列可能会被其它线程修改,所以这个值是近似值)
empty()                                                   如果队列为空返回 True,否则返回 False
full()                                                        如果队列已满返回 True,否则返回 False
put(item,block=0)                                   把 item 放到队列中,如果给了 block(不为 0),函数会一直阻塞到队列中有空间为止
get(block=0)                                           从队列中取一个对象,如果给了 block(不为 0),函数会一直阻塞到队列中有对象为止

import threading
from Queue import Queue
from time import sleep,ctime
import random


class MyThread(threading.Thread):　　#threading另一种用法：继承threading.Thread类，覆写run函数
    def __init__(self,func,args,name=''):
        threading.Thread.__init__(self)
        self.name=name
        self.func=func
        self.args=args
        
    def run(self):
        print "start "+self.name+" at;"+ctime()
        apply(self.func,self.args)
        print "stop "+self.name+" at;"+ctime()
    
def writeQ(queue):
    queue.put("item",1)
    print "product object for Q. now size: ",queue.qsize()
    
def readQ(queue):
    queue.get(1)
    print "consumed object from Q. now size:",queue.qsize()
    
def writer(queue,n):
    for i in range(n):
        writeQ(queue)
        sleep(random.randint(1,3))
        
def reader(queue,n):
    for i in range(n):
        readQ(queue)
        sleep(random.randint(3,5))
        
        
funcs=[writer,reader]
nfuncs=len(funcs)

def main():
    queue=Queue(15)
    n=random.randint(4,7)
    thread=[]
    
    for i in range(nfuncs):
        t=MyThread(funcs[i],args=(queue,n),name=funcs[i].__name__)
        thread.append(t)
        
    for i in range(nfuncs):
        thread[i].start()
        
    for i in range(nfuncs):
        thread[i].join()
        
    print "ALL DONE"
if __name__ == '__main__':
    main()

运行结果：

start writer at;Tue Jun 13 21:58:23 2017
 start reader at;Tue Jun 13 21:58:23 2017
product object for Q. now size: 0
 consumed object from Q. now size:1
 
product object for Q. now size:  1
consumed object from Q. now size: 0
product object for Q. now size:  1
product object for Q. now size:  2
product object for Q. now size:  3
consumed object from Q. now size: 2
product object for Q. now size:  3
product object for Q. now size:  4
consumed object from Q. now size: 3
stop writer at;Tue Jun 13 21:58:37 2017
consumed object from Q. now size: 2
consumed object from Q. now size: 1
consumed object from Q. now size: 0
stop reader at;Tue Jun 13 21:58:50 2017
ALL DONE

6.线程池

线程池采用预创建的技术，在应用程序启动之后，将立即创建一定数量的线程(N1)，放入空闲队列中。当任务到来后，缓冲池选择一个空闲线程，把任务传入此线程中运行。当N1个线程都在处理任务后，缓冲池自动创建一定数量的新线程，用于处理更多的任务。在任务执行完毕后线程也不退出，而是继续保持在池中等待下一次的任务。当系统比较空闲时，大部分线程都一直处于暂停状态，线程池自动销毁一部分线程，回收系统资源。

一般线程池都必须具备下面几个组成部分：
　　线程池管理器:用于创建并管理线程池
　　工作线程: 线程池中实际执行的线程
　　任务接口: 尽管线程池大多数情况下是用来支持网络服务器，但是我们将线程执行的任务抽象出来，形成任务接口，从而是的线程池与具体的任务无关。
　　任务队列:线程池的概念具体到实现则可能是队列，链表之类的数据结构，其中保存执行线程。

　　我们把任务放进队列中去，然后开N个线程，每个线程都去队列中取一个任务，执行完了之后告诉系统说我执行完了，然后接着去队列中取下一个任务，直至队列中所有任务取空，退出线程。

进程池详解：http://www.cnblogs.com/goodhacker/p/3359985.html

将生产者消费者问题改写成进程池：

import Queue
import threading
from time import sleep, ctime,time
import random


class wordManage( object ):　　　　#线程池管理器
    def __init__(self, func, args, work_num=100, thread_num=2):
        self.work_queue = Queue.Queue()
        self.func = func
        self.args = args
        self.thread = []
        self.__init_work_queue( work_num )
        self.__init_thread_pool( thread_num )
    
    def __init_thread_pool(self, thread_num):　　#初始化线程池
        for i in range( thread_num ):
            self.thread.append( Work( self.work_queue ) )
    
    def __init_work_queue(self, work_num):　　　　#初始化工作队列
        for i in range( work_num ):
            self.add_job()
    
    def add_job(self):　　　　　　　　　　#添加工作
        self.work_queue.put( (self.func, self.args) )
    
    def wait_allcomplete(self):
        for item in self.thread:
            if item.isAlive:
                item.join()


class Work( threading.Thread ):
    def __init__(self, work_queue):
        threading.Thread.__init__( self )
        self.work_queue = work_queue
        self.start()
    
    def run(self):
        while True:
            try:
                func, args = self.work_queue.get( block=False )
                apply(func,args)　　　　　　　　　　
                self.work_queue.task_done()
            except:
                break


def writeQ(queue):
    queue.put( "item", 1 )
    print "product object for Q. now size: ", queue.qsize()


def readQ(queue):
    queue.get( 1 )
    print "consumed object from Q. now size:", queue.qsize()


def writer(queue, n):
    for i in range( n ):
        writeQ( queue )
        sleep( random.randint( 1, 3 ) )


def reader(queue, n):
    for i in range( n ):
        readQ( queue )
        sleep( random.randint( 3, 5 ) )



def main():
    print "work start at "+ctime()
    queue = Queue.Queue( 15 )
    n = random.randint( 1, 3 )
    write_manage = wordManage( func=writer, args=(queue, n), work_num=10, thread_num=3 )
    read_manage = wordManage( func=reader, args=(queue, n), work_num=10, thread_num=3 )
    write_manage.wait_allcomplete()
    read_manage.wait_allcomplete()
    print "ALL DONE AT "+ctime()

if __name__ == '__main__':
    main()