线程

多进程和多线程选择

    一般来说IO密集型用线程(python中线程无法夸cpu执行，i在IO操作只会少许占用cpu，线程安全)，计算密集型用多进程(多进程可以跑在不同cpu上)。线程带GIL锁，同时操作IO时候只有一个可以成功操作。

    多进程和线程开多少合适？ 理论上多进程个数可以等于cpu核数，线程数量看业务而定，cpu在线程上切换也会占用时间(上下文切换)

thread方法说明

t.start() : 激活线程，

t.getName() : 获取线程的名称

t.setName() ： 设置线程的名称 

t.name : 获取或设置线程的名称

t.is_alive() ： 判断线程是否为激活状态

t.isAlive() ：判断线程是否为激活状态

t.setDaemon() 设置为后台线程或前台线程（默认：False）;通过一个布尔值设置线程是否为守护线程，必须在执行start()方法之后才可以使用。如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止；如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止

t.isDaemon() ： 判断是否为守护线程

t.ident ：获取线程的标识符。线程标识符是一个非零整数，只有在调用了start()方法之后该属性才有效，否则它只返回None。

t.join() ：逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义

t.run() ：线程被cpu调度后自动执行线程对象的run方法

简单线程操作

#-*- coding:utf-8 -*-
import time
import threading


def f0():pass


def f1(*args):
    time.sleep(10)
    f0()

#线程执行前准备
t= threading.Thread(target=f1,args=(111,222))
#开始执行线程
t.start()


t= threading.Thread(target=f1,args=(111,222))
t.start()
t= threading.Thread(target=f1,args=(111,222))
t.start()

线程锁

我们使用线程对数据进行操作的时候，如果多个线程同时修改某个数据，可能会出现不可预料的结果，为了保证数据的准确性，引入了锁的概念。

#-*- coding:utf-8  -*-
import threading
import time
 
globals_num = 0
 
lock = threading.RLock()
 
def Func():
    lock.acquire()  # 获得锁 
    global globals_num
    globals_num += 1
    time.sleep(1)
    print(globals_num)
    lock.release()  # 释放锁 
 
for i in range(10):
    t = threading.Thread(target=Func)
    t.start()

threading.RLock和threading.Lock 的区别

RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。 如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的琐。

线程之间通知(Event)

Event是线程间通信最间的机制之一：一个线程发送一个event信号，其他的线程则等待这个信号。用于主线程控制其他线程的执行。 Events 管理一个flag，这个flag可以使用set()设置成True或者使用clear()重置为False，wait()则用于阻塞，在flag为True之前。flag默认为False。

• Event.wait([timeout]) ： 堵塞线程，直到Event对象内部标识位被设为True或超时（如果提供了参数timeout）。
• Event.set() ：将标识位设为Ture
• Event.clear() ： 将标识位设为False。
• Event.isSet() ：判断标识位是否为Ture。

#-*- coding:utf-8  -*-
import threading

def do(event):
    print('start')
    #线程在这里阻塞了。等待event_obj.set()方法才能继续执行线程
    event.wait()
    print('execute')

event_obj = threading.Event()
for i in range(10):#加锁
    t = threading.Thread(target=do,args=(event_obj,))
    t.start()

#event_obj.clear()设置为flase 线程就阻塞了。
event_obj.clear()

inp = raw_input('input:')#放锁
print(inp)
if inp == 'true':
    #发送event事件，线程有阻塞变成无阻塞
    event_obj.set()

当某事过程完毕后改变事件，被阻塞的线程将继续执行

threading.Condition:

一个condition变量总是与某些类型的锁相联系，这个可以使用默认的情况或创建一个，当几个condition变量必须共享和同一个锁的时候，是很有用的。锁是conditon对象的一部分：没有必要分别跟踪。

condition变量服从上下文管理协议：with语句块封闭之前可以获取与锁的联系。 acquire() 和 release() 会调用与锁相关联的相应的方法。

其他和锁关联的方法必须被调用，wait()方法会释放锁，当另外一个线程使用 notify() or notify_all()唤醒它之前会一直阻塞。一旦被唤醒，wait()会重新获得锁并返回，

Condition类实现了一个conditon变量。 这个conditiaon变量允许一个或多个线程等待，直到他们被另一个线程通知。 如果lock参数，被给定一个非空的值，，那么他必须是一个lock或者Rlock对象，它用来做底层锁。否则，会创建一个新的Rlock对象，用来做底层锁。

• wait(timeout=None) ： 等待通知，或者等到设定的超时时间。当调用这wait()方法时，如果调用它的线程没有得到锁，那么会抛出一个RuntimeError 异常。 wati()释放锁以后，在被调用相同条件的另一个进程用notify() or notify_all() 叫醒之前 会一直阻塞。wait() 还可以指定一个超时时间。

如果有等待的线程，notify()方法会唤醒一个在等待conditon变量的线程。notify_all() 则会唤醒所有在等待conditon变量的线程。

注意： notify()和notify_all()不会释放锁，也就是说，线程被唤醒后不会立刻返回他们的wait() 调用。除非线程调用notify()和notify_all()之后放弃了锁的所有权。

在典型的设计风格里，利用condition变量用锁去通许访问一些共享状态，线程在获取到它想得到的状态前，会反复调用wait()。修改状态的线程在他们状态改变时调用 notify() or notify_all()，用这种方式，线程会尽可能的获取到想要的一个等待者状态。 例子： 生产者-消费者模型

#-*- coding:utf-8 -*-
import threading
import time


def consumer(cond):
    #不管释放锁还是加锁都需要先用with
    with cond:
        print("consumer before wait")
        #等待解锁,如果没写cond.wait()这个方法程序会往下执行
        cond.wait()
        print("consumer after wait")


#释放锁
def producer(cond):
    with cond:
        print("producer before notifyAll")
        #释放锁，这个必须是同一个condition对象
        cond.notifyAll()
        print("producer after notifyAll")

#定义condition对象
condition = threading.Condition()

c1 = threading.Thread(name="c1", target=consumer, args=(condition,))
c2 = threading.Thread(name="c2", target=consumer, args=(condition,))

p = threading.Thread(name="p", target=producer, args=(condition,))

c1.start()
time.sleep(2)
c2.start()
time.sleep(2)

#释放锁
p.start()

自定义线程池(线程池默认没有)

一简单的方法实现线程池

#-*- coding:utf-8 -*-
import  threading
import time
import Queue


#线程池
class ThreadPool(object):

    def __init__(self,max_num=20):
        #创建队列默认20个
        self.queue = Queue.Queue(max_num)

        #增加线程方法到Queue里面
        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)


def func(pool,num):
    time.sleep(1)
    print(num)
    #取一个线程就需要往队列中在放一个线程
    pool.add_thread()

p = ThreadPool(5)

for i in range(10):
    thread=p.get_thread()
    t= thread(target=func,args=(p,i,))
    t.start()

二增强版线程池

#-*- coding:utf-8 -*-
import  threading
import time
import Queue
import contextlib

StopEvent=object()

class ThreadPool(object):
    def __init__(self,max_num):
        #这个Q是无限大的用来接受任务
        self.q = Queue.Queue()

        #执行线程计数器
        self.work_num =0

        #为True线程会立即终止执行
        self.terminal = False

        #最多创建的线程数
        self.max_num = max_num

        #真实创建的线程列表
        self.generate_list = []

        #空闲线程列表
        self.free_list = []

    def run(self,func,args,cllback=None):

        #w元组类型存储的执行函数，参数，callback函数。put到任务队列中去
        w = (func,args,cllback)
        self.q.put(w)

        #判断是否创建新的线程
        #1空线程等于0
        #2工作线程小于最大线程数
        if len(self.free_list) == 0 and len(self.generate_list)<self.max_num:
           #创建线程
           self.generate_thread()

    #创建线程
    def generate_thread(self):
        #启动一个线程然后调用self.call方法
        t = threading.Thread(target=self.call)
        t.start()

    #当任务执行完毕后往队列中增加停止符
    def close(self):
        while self.generate_list:
            self.q.put(StopEvent)


    def terminate(self):
        self.terminal = True
        self.close()
        #清空队列.这里已经终止队列中可能还有很多任务没有完成
        self.q.empty()

    #做任务
    def call(self):
        #获取当前线程池
        current_thread = threading.currentThread
        #加入工作列表
        self.generate_list.append(current_thread)

        #获取q中的任务,没有任务get就会阻塞线程
        event = self.q.get()

        #增加计数器
        self.work_num +=1

        while event != StopEvent:
             #获取的event 如果是元组那么它就是任务,其它格式肯定不是任务
             #解开任务包
             func, args ,callable = event


             try:
                    #获取执行方法的返回值
                    result = func(*args)
                    #状态
                    status = True
             except Exception as e:
                    #func函数执行过程中报异常了，这里捕获一下
                    result = e
                    #状态改为false
                    status = False

             #判断一下是否需要回调
             if callable is not None:
                 try:
                     #执行回调函数,并且把结果当参数传递到callable中
                     #回调函数中只需要判断下status就知道func函数是否执行成功
                     callable(status,result)
                 except Exception as e:
                     pass
             if self.terminal:
                 event = StopEvent
             else:
                 #标记，free_list空闲列表中
                 self.free_list.append(current_thread)

                 #获取任务，如果没有get方法将阻塞线程，等待任务到来
                 event = self.q.get()

                 #从空闲队列中移除线程
                 self.free_list.remove(current_thread)

        else:#这里event == StopEvent 就执行这里，说明event不是元组
            self.generate_list.remove(current_thread)


def f1(args):
    import time
    time.sleep(0.5)
    print(args)

#线程池初始线程设置
pool = ThreadPool(10)

for i in range(5):
    pool.run(f1,(i,))

#增加停止符
#pool.close()

#立即停止
pool.terminate()

#查看用了多少个工作线程
time.sleep(3)
print('
work:%d'%pool.work_num)

用with改造下面流程：

self.free_list.append(current_thread)
event = self.q.get()
self.free_list.remove(current_thread)

增加ThreadPool类中的一个方法：

#上下文管理函数
    @contextlib.contextmanager
    def work_status(self,current_thread):
        self.free_list.append(current_thread)
        try:
              yield
        finally:
            self.free_list.remove(current_thread)

将上面改造的三行代码改成：

with self.work_status(current_thread):
                     event = self.q.get()