学习PYTHON之路， DAY 10 进程、线程、协程篇

线程

线程是应用程序中工作的最小单元。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。

直接调用

import threading
import time

def show(arg):
    time.sleep(1)
    print 'thread'+str(arg)

for i in range(10):
    t = threading.Thread(target=show, args=(i,))
    t.start()

print 'main thread stop'

继承式调用

import threading
import time

class MyThread(threading.Thread):
    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):#定义每个线程要运行的函数

        print("running on number:%s" %self.num)

        time.sleep(3)

if __name__ == '__main__':

    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

更多方法：

• start            线程准备就绪，等待CPU调度
• setName      为线程设置名称
• getName      获取线程名称
• setDaemon   设置为后台线程或前台线程（默认）
                     如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止
                      如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
• join              逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
• run              线程被cpu调度后自动执行线程对象的run方法

 import time
 import threading

 def run(n):

     print('[%s]------running----\n' % n)
     time.sleep(2)
     print('--done--')

 def main():
     for i in range(5):
         t = threading.Thread(target=run,args=[i,])
         t.start()
         t.join(1)
         print('starting thread', t.getName())

 m = threading.Thread(target=main,args=[])
 m.setDaemon(True) #将main线程设置为Daemon线程,它做为程序主线程的守护线程,当主线程退出时,m线程也会退出,由m启动的其它子线程会同时退出,不管是否执行完任务
 m.start()
 m.join(timeout=2)
 print("---main thread done----")

守护线程

线程锁（Lock、RLock）

由于线程之间是进行随机调度，并且每个线程可能只执行n条执行之后，当多个线程同时修改同一条数据时可能会出现脏数据，所以，出现了线程锁 - 同一时刻允许一个线程执行操作。

import threading
import time

gl_num = 0

lock = threading.RLock()

def Func():
    lock.acquire()
    global gl_num
    gl_num +=1
    time.sleep(1)
    print gl_num
    lock.release()

for i in range(10):
    t = threading.Thread(target=Func)
    t.start()

信号量（Semaphore）

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

 import threading,time

 def run(n):
     semaphore.acquire()
     time.sleep(1)
     print("run the thread: %s" %n)
     semaphore.release()

 if __name__ == '__main__':

     num= 0
     semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
     for i in range(20):
         t = threading.Thread(target=run,args=(i,))
         t.start()

事件（event）

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

• clear：将“Flag”设置为False
• set：将“Flag”设置为True

import threading

def do(event):
    print 'start'
    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()
inp = raw_input('input:')
if inp == 'true':
    event_obj.set()

条件（Condition）

使得线程等待，只有满足某条件时，才释放n个线程

def condition_func():

    ret = False
    inp = input('>>>')
    if inp == '':
        ret = True

    return ret

def run(n):
    con.acquire()
    con.wait_for(condition_func)
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

队列

class queue.Queue(maxsize=0) #先入先出

class queue.LifoQueue(maxsize=0) #last in fisrt out 

class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列

Queue.qsize()

Queue.empty() #return True if empty

Queue.full() # return True if full

Queue.put(item, block=True, timeout=None)Queue.put_nowait(item)Equivalent to put(item, False).

Queue.get(block=True, timeout=None)Queue.get_nowait()

Equivalent to get(False)

Queue.task_done()

 import time,random
 import queue,threading
 q = queue.Queue()
 def Producer(name):
   count = 0
   while count <20:
     time.sleep(random.randrange(3))
     q.put(count)
     print('Producer %s has produced %s baozi..' %(name, count))
     count +=1
 def Consumer(name):
   count = 0
   while count <20:
     time.sleep(random.randrange(4))
     if not q.empty():
         data = q.get()
         print(data)
         print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
     else:
         print("-----no baozi anymore----")
     count +=1
 p1 = threading.Thread(target=Producer, args=('A',))
 c1 = threading.Thread(target=Consumer, args=('B',))
 p1.start()
 c1.start()

线程池

 import queue
 import threading
 import time

 class ThreadPool:
     def __init__(self, maxsize=5):
         self.maxsize = maxsize
         self._q = queue.Queue(maxsize)
         for i in range(maxsize):
             self._q.put(threading.Thread)
         # 【threading.Thread，threading.Thread，threading.Thread，threading.Thread，threading.Thread】
     def get_thread(self):
         return self._q.get()

     def add_thread(self):
         self._q.put(threading.Thread)

 pool = ThreadPool(5)

 def task(arg,p):
     print(arg)
     time.sleep(1)
     p.add_thread()

 for i in range(100):
     # threading.Thread类
     t = pool.get_thread()
     obj = t(target=task,args=(i,pool,))
     obj.start()

简单版

 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
 # Author:Alex Li
 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.empty()

     @contextlib.contextmanager
     def worker_state(self, state_list, worker_thread):
         """
         用于记录线程中正在等待的线程数
         """
         state_list.append(worker_thread)
         try:
             yield
         finally:
             state_list.remove(worker_thread)

 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(300):
     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))

复杂版

上下文管理

 import contextlib

 @contextlib.contextmanager #加了这个装饰器，可以用with
 def work(free_list, worker_thread):
     free_list.append(worker_thread)
     try:
         yield
     finally:
         free_list.remove(worker_thread)

 free_list = []
 worker_thread = ''
 with work(free_list, worker_thread):
     print(123)

运行顺序

多进程

from multiprocessing import Process
import time
def f(name):
    time.sleep(2)
    print('hello', name)

if __name__ == '__main__': #在windos进程只能做测试，一定要写这句
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

进程间通讯

Queues

使用方法跟threading里的queue差不多

from multiprocessing import Process, Queue

def f(i,q):
    print(i,q.get())

if __name__ == '__main__':
    q = Queue()

    q.put("h1")
    q.put("h2")
    q.put("h3")

    for i in range(10):
        p = Process(target=f, args=(i,q,))
        p.start()

Managers

 from multiprocessing import Process, Manager

 def f(d, l):
     d[1] = ''
     d[''] = 2
     d[0.25] = None
     l.append(1)
     print(l)

 if __name__ == '__main__':
     with Manager() as manager:
         d = manager.dict()

         l = manager.list(range(5))
         p_list = []
         for i in range(10):
             p = Process(target=f, args=(d, l))
             p.start()
             p_list.append(p)
         for res in p_list:
             res.join()

         print(d)
         print(l)

协程

协程一个标准定义：

1. 必须在只有一个单线程里实现并发
2. 修改共享数据不需加锁
3. 用户程序里自己保存多个控制流的上下文栈
4. 一个协程遇到IO操作自动切换到其它协程

greenlet

 from greenlet import greenlet

 def test1():
     print 12
     gr2.switch()
     print 34
     gr2.switch()

 def test2():
     print 56
     gr1.switch()
     print 78

 gr1 = greenlet(test1)
 gr2 = greenlet(test2)
 gr1.switch()

gevent

 import gevent

 def foo():
     print('Running in foo')
     gevent.sleep(0)
     print('Explicit context switch to foo again')

 def bar():
     print('Explicit context to bar')
     gevent.sleep(0)
     print('Implicit context switch back to bar')

 gevent.joinall([
     gevent.spawn(foo),
     gevent.spawn(bar),
 ])