Python线程

Threading用于提供线程相关的操作，线程是应用程序中工作的最小单元。

threading模块对象

例子：

import threading
import time
 
def sayhi(num): #定义每个线程要运行的函数
 
    print("running on number:%s" %num)
 
    time.sleep(3)
 
if __name__ == ‘__main__‘:
 
    t1 = threading.Thread(target=sayhi,args=(1,)) #生成一个线程实例
    t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一个线程实例
 
    t1.start() #启动线程
    t2.start() #启动另一个线程
 
    print(t1.getName()) #获取线程名
    print(t2.getName())

上述代码创建了10个“前台”线程，然后控制器就交给了CPU，CPU根据指定算法进行调度，分片执行指令。

更多方法：

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

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

线程锁（Lock、RLock）

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

请求锁定 — 进入锁定池等待 — 获取锁 — 已锁定 — 释放锁
Lock（指令锁）是可用的最低级的同步指令。Lock处于锁定状态时，不被特定的线程拥有。Lock包含两种状态——锁定和非锁定，以及两个基本的方法。
可以认为Lock有一个锁定池，当线程请求锁定时，将线程至于池中，直到获得锁定后出池。池中的线程处于状态图中的同步阻塞状态。
构造方法：
Lock()
实例方法：
acquire([timeout]): 使线程进入同步阻塞状态，尝试获得锁定。
release(): 释放锁。使用前线程必须已获得锁定，否则将抛出异常。

没有锁

import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print(‘--get num:‘,num )
    time.sleep(1)
    num  -=1 #对此公共变量进行-1操作
 
num = 100  #设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
 
print(‘final num:‘, num )

正常来讲，这个num结果应该是0， 但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？ 哈，很简单，假设你有A,B两个线程，此时都 要对num 进行减1操作， 由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。那怎么办呢？ 很简单，每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。 

*注：不要在3.x上运行，不知为什么，3.x上的结果总是正确的，可能是自动加了锁

加锁版本

import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print(‘--get num:‘,num )
    time.sleep(1)
    lock.acquire() #修改数据前加锁
    num  -=1 #对此公共变量进行-1操作
    lock.release() #修改后释放
 
num = 100  #设定一个共享变量
thread_list = []
lock = threading.Lock() #生成全局锁
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
print(‘final num:‘, num )

RLock（递归锁）

RLock（可重入锁）是一个可以被同一个线程请求多次的同步指令。RLock使用了“拥有的线程”和“递归等级”的概念，处于锁定状态时，RLock被某个线程拥有。拥有RLock的线程可以再次调用acquire()，释放锁时需要调用release()相同次数。
可以认为RLock包含一个锁定池和一个初始值为0的计数器，每次成功调用 acquire()/release()，计数器将+1/-1，为0时锁处于未锁定状态。
构造方法：
RLock()

说白了就是在一个大锁中还要再包含子锁

import threading,time
 
def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num +=1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global  num2
    num2+=1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print(‘--------between run1 and run2-----‘)
    res2 = run2()
    lock.release()
    print(res,res2)
 
 
if __name__ == ‘__main__‘:
 
    num,num2 = 0,0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()
 
while threading.active_count() != 1:
    print(threading.active_count())
else:
    print(‘----all threads done---‘)
    print(num,num2)

信号量（Semaphore）

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

import threading,time
 
def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s\n" %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()
 
while threading.active_count() != 1:
    pass #print threading.active_count()
else:
    print(‘----all threads done---‘)
    print(num)

事件（event）

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

Event内部包含了一个标志位，初始的时候为false。
可以使用使用set()来将其设置为true；
或者使用clear()将其从新设置为false；
可以使用is_set()来检查标志位的状态；
另一个最重要的函数就是wait(timeout=None)，用来阻塞当前线程，直到event的内部标志位被设置为true或者timeout超时。如果内部标志位为true则wait()函数理解返回。

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

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

 通过Event来实现两个或多个线程间的交互，下面是一个红绿灯的例子，即起动一个线程做交通指挥灯，生成几个线程做车辆，车辆行驶按红灯停，绿灯行的规则。

import threading,time
import random
def light():
    if not event.isSet():
        event.set() #wait就不阻塞 #绿灯状态
    count = 0
    while True:
        if count < 10:
            print(‘\033[42;1m--green light on---\033[0m‘)
        elif count <13:
            print(‘\033[43;1m--yellow light on---\033[0m‘)
        elif count <20:
            if event.isSet():
                event.clear()
            print(‘\033[41;1m--red light on---\033[0m‘)
        else:
            count = 0
            event.set() #打开绿灯
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(random.randrange(10))
        if  event.isSet(): #绿灯
            print("car [%s] is running.." % n)
        else:
            print("car [%s] is waiting for the red light.." %n)
if __name__ == ‘__main__‘:
    event = threading.Event()
    Light = threading.Thread(target=light)
    Light.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

 这里还有一个event使用的例子，员工进公司门要刷卡， 我们这里设置一个线程是“门”， 再设置几个线程为“员工”，员工看到门没打开，就刷卡，刷完卡，门开了，员工就可以通过。

#_*_coding:utf-8_*_
__author__ = ‘Alex Li‘
import threading
import time
import random

def door():
    door_open_time_counter = 0
    while True:
        if door_swiping_event.is_set():
            print("\033[32;1mdoor opening....\033[0m")
            door_open_time_counter +=1

        else:
            print("\033[31;1mdoor closed...., swipe to open.\033[0m")
            door_open_time_counter = 0 #清空计时器
            door_swiping_event.wait()


        if door_open_time_counter > 3:#门开了已经3s了,该关了
            door_swiping_event.clear()

        time.sleep(0.5)


def staff(n):

    print("staff [%s] is comming..." % n )
    while True:
        if door_swiping_event.is_set():
            print("\033[34;1mdoor is opened, passing.....\033[0m")
            break
        else:
            print("staff [%s] sees door got closed, swipping the card....." % n)
            print(door_swiping_event.set())
            door_swiping_event.set()
            print("after set ",door_swiping_event.set())
        time.sleep(0.5)
door_swiping_event  = threading.Event() #设置事件


door_thread = threading.Thread(target=door)
door_thread.start()



for i in range(5):
    p = threading.Thread(target=staff,args=(i,))
    time.sleep(random.randrange(3))
    p.start()

条件（Condition）

可以把Condition理解为一把高级的琐，它提供了比Lock, RLock更高级的功能，允许我们能够控制复杂的线程同步问题。threadiong.Condition在内部维护一个琐对象（默认是RLock），可以在创建Condigtion对象的时候把琐对象作为参数传入。Condition也提供了acquire, release方法，其含义与琐的acquire, release方法一致，其实它只是简单的调用内部琐对象的对应的方法而已。Condition还提供了如下方法(特别要注意：这些方法只有在占用琐(acquire)之后才能调用，否则将会报RuntimeError异常。)：

Condition.wait([timeout]):
wait方法释放内部所占用的琐，同时线程被挂起，直至接收到通知被唤醒或超时（如果提供了timeout参数的话）。当线程被唤醒并重新占有琐的时候，程序才会继续执行下去。

Condition.notify():
唤醒一个挂起的线程（如果存在挂起的线程）。注意：notify()方法不会释放所占用的琐。

Condition.notify_all()
Condition.notifyAll()
唤醒所有挂起的线程（如果存在挂起的线程）。注意：这些方法不会释放所占用的琐。

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

import threading
 
def run(n):
    con.acquire()
    con.wait()
    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()
 
    while True:
        inp = input(‘>>>‘)
        if inp == ‘q‘:
            break
        con.acquire()
        con.notify(int(inp))
        con.release()

def condition_func():

    ret = False
    inp = input(‘>>>‘)
    if inp == ‘1‘:
        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()

Timer

定时器，指定n秒后执行某操作

def hello():
    print("hello, world")
 
t = Timer(30.0, hello)
t.start()  # after 30 seconds, "hello, world" will be printed

queue队列

初识Queue模块

Queue 模块实现了多生产者、多消费者队列。它特别适用于信息必须在多个线程间安全地交换的多线程程序中。这个模块中的 Queue 类实现了所有必须的锁语义。它依赖于 Python 中的线程支持的可用性；参见threading 模块。

模块实现了三类队列，主要差别在于取得数据的顺序上。在FIFO（First In First Out，先进先出）队列中，最早加入的任务会被最先得到。在LIFO（Last In First Out，后进先出）队列中，最后加入的任务会被最先得到（就像栈一样）。在优先队列中，任务被保持有序（使用heapq模块），拥有最小值的任务（优先级最高）被最先得到。

模块实现了三类队列：FIFO（First In First Out，先进先出，默认为该队列）、LIFO（Last In First Out，后进先出）、基于优先级的队列。以下为其常用方法：

    先进先出  q = Queue.Queue(maxsize)
    后进先出  a = Queue.LifoQueue(maxsize)
    优先级  Queue.PriorityQueue(maxsize)
    Queue.qsize() 返回队列的大小
    Queue.empty() 如果队列为空，返回True,反之False
    Queue.full() 如果队列满了，返回True,反之False
    Queue.full 与 maxsize 大小对应
    Queue.put(item) 写入队列，timeout等待时间   非阻塞
    Queue.get([block[, timeout]]) 获取队列，timeout等待时间
    Queue.get_nowait() 相当Queue.get(False)
    Queue.put_nowait(item) 相当Queue.put(item, False)
    Queue.task_done() 在完成一项工作之后，函数向任务已经完成的队列发送一个信号
    Queue.join()： 实际上意味着等到队列为空，再执行别的操作 

Queue 模块定义了下列的类和异常：

class Queue.Queue(maxsize=0)

构造一个FIFO队列。maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。定义队列时有一个默认的参数maxsize, 如果不指定队列的长度，即manxsize=0,那么队列的长度为无限长，如果定义了大于0的值，那么队列的长度就是maxsize。

class Queue.LifoQueue(maxsize=0)

构造一个LIFO队列。maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。

出现于版本2.6.

class Queue.PriorityQueue(maxsize=0)

构造一个优先队列。maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。

拥有最小值的任务会被最先得到（sorted(list(entries))[0]的返回值即为拥有最小值的任务）。任务的典型模式就是如(priority_number, data)这样的元组。

出现于版本2.6.

exception Queue.Empty

在空的Queue对象上调用非阻塞的get()（或者get_nowait()）会抛出此异常。

exception Queue.Full

在满的Queue对象上调用非阻塞的put()（或者put_nowait()）会抛出此异常。

 Queue 对象

Queue对象（Queue、LifoQueue和PriorityQueue）提供了下述的公共方法。

Queue.qsize()

返回队列的近似大小。注意，队列大小大于0并不保证接下来的get()调用不会被阻塞，队列大小小于maxsize也不保证接下来的put()调用不会被阻塞。

Queue.empty()

如果队列为空返回True，否则返回False。如果empty()返回True并不保证接下来的put()调用不会被阻塞。类似的，如果empty()返回False也不能保证接下来的get()调用不会被阻塞。

Queue.full()

如果队列是满的返回True，否则返回False。如果full()返回True并不能保证接下来的get()调用不会被阻塞。类似的，如果full()返回False并不能保证接下来的put()调用不会被阻塞。

Queue.put(item[, block[, timeout]])

将item放入队列中。如果可选的参数block为真且timeout为空对象（默认的情况，阻塞调用，无超时），如有必要（比如队列满），阻塞调用线程，直到有空闲槽可用。如果timeout是个正整数，阻塞调用进程最多timeout秒，如果一直无空闲槽可用，抛出Full异常（带超时的阻塞调用）。如果block为假，如果有空闲槽可用将数据放入队列，否则立即抛出Full异常（非阻塞调用，timeout被忽略）。

出现于版本2.3： timeout参数。

Queue.put_nowait(item)

等同于put(item, False)（非阻塞调用）。

Queue.get([block[, timeout]])

从队列中移除并返回一个数据。如果可选的参数block为真且timeout为空对象（默认的情况，阻塞调用，无超时），阻塞调用进程直到有数据可用。如果timeout是个正整数，阻塞调用进程最多timeout秒，如果一直无数据可用，抛出Empty异常（带超时的阻塞调用）。如果block为假，如果有数据可用返回数据，否则立即抛出Empty异常（非阻塞调用，timeout被忽略）。

出现于版本2.3： timeout参数。

Queue.get_nowait()

等同于get(False)（非阻塞调用）。

为了跟踪入队任务被消费者线程完全的处理掉，Queue对象提供了两个额外的方法。

Queue.task_done()

意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来的task_done()调用告诉队列该任务已经处理完毕。

如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。

如果该方法被调用的次数多于被放入队列中的任务的个数，ValueError异常会被抛出。

出现于版本2.5。

Queue.join()

阻塞调用线程，直到队列中的所有任务被处理掉。

只要有数据被加入队列，未完成的任务数就会增加。当消费者线程调用task_done()（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，join()解除阻塞。

出现于版本2.5。

import queue

# q = queue.Queue(2)
#
# q.put(1)
# q.put(2)
# #q.put(3)
# #q.put_nowait(3)
#
# print(q.empty())
# print(q.full())
#
# print(q.get())
# print(q.get())

权重：
q = queue.PriorityQueue()
# q.put((3,‘aaaaa‘))
# q.put((3,‘bbbbb‘))
# q.put((1,‘ccccc‘))
# q.put((3,‘ddddd‘))
q.put([6,‘alex‘])
q.put([3,‘jack‘])
q.put([5,‘rain‘])
print(q.get())
print(q.get())

GIL，全局解释器锁
总结：
多进程，多线程，提供并发
IO密集型：多线程
计算密集型：多进程

GIL是什么

首先需要明确的一点是 GIL 并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。就好比C++是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把 GIL 归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL

那么CPython实现中的GIL又是什么呢？GIL全称 Global Interpreter Lock 为了避免误导，我们还是来看一下官方给出的解释：

好吧，是不是看上去很糟糕？一个防止多线程并发执行机器码的一个Mutex，乍一看就是个BUG般存在的全局锁嘛！别急，我们下面慢慢的分析。

为什么会有GIL

由于物理上得限制，各CPU厂商在核心频率上的比赛已经被多核所取代。为了更有效的利用多核处理器的性能，就出现了多线程的编程方式，而随之带来的就是线程间数据一致性和状态同步的困难。 即使在CPU内部的Cache也不例外 ，为了有效解决多份缓存之间的数据同步时各厂商花费了不少心思，也不可避免的带来了一定的性能损失。

Python当然也逃不开，为了利用多核，Python开始支持多线程。 而解决多线程之间数据完整性和状态同步的最简单方法自然就是加锁。 于是有了GIL这把超级大锁，而当越来越多的代码库开发者接受了这种设定后，他们开始大量依赖这种特性（即默认python内部对象是thread-safe的，无需在实现时考虑额外的内存锁和同步操作）。

慢慢的这种实现方式被发现是蛋疼且低效的。但当大家试图去拆分和去除GIL的时候，发现大量库代码开发者已经重度依赖GIL而非常难以去除了。有多难？做个类比，像MySQL这样的“小项目”为了把Buffer Pool Mutex这把大锁拆分成各个小锁也花了从5.5到5.6再到5.7多个大版为期近5年的时间，本且仍在继续。MySQL这个背后有公司支持且有固定开发团队的产品走的如此艰难，那又更何况Python这样核心开发和代码贡献者高度社区化的团队呢？

所以简单的说GIL的存在更多的是历史原因。如果推到重来，多线程的问题依然还是要面对，但是至少会比目前GIL这种方式会更优雅。

GIL的影响

从上文的介绍和官方的定义来看，GIL无疑就是一把全局排他锁。毫无疑问全局锁的存在会对多线程的效率有不小影响。甚至就几乎等于Python是个单线程的程序。那么读者就会说了，全局锁只要释放的勤快效率也不会差啊。只要在进行耗时的IO操作的时候，能释放GIL，这样也还是可以提升运行效率的嘛。或者说再差也不会比单线程的效率差吧。理论上是这样，而实际上呢？Python比你想的更糟。

总结

生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

下面来学习一个最基本的生产者消费者模型的例子

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

import threading
import queue
import time

def consumer(name):

    while True:
        print(‘%s 取到骨头[%s]并吃了‘ % (name,q.get()))
        time.sleep(0.5)
        q.task_done()

def producer(name):
    count = 0
    #while q.qsize() <5:
    for i in range(10):
        print(‘%s生成了骨头‘ %name,count)
        q.put(count)
        count += 1
        time.sleep(0.3)
    q.join()
    print("------吃完了-------")

#生成一个队列
q = queue.Queue(maxsize=4)

#生成两个线程
p = threading.Thread(target=producer,args=(‘Tom‘,))
#p2 = threading.Thread(target=producer,args=(‘Tom‘,))
c = threading.Thread(target=consumer,args=(‘Jack‘,))
p.start()
#p2.start()
c.start()

 第二个：

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 = http://www.mamicode.com/q.get()"-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=(‘A‘,))
c1 = threading.Thread(target=Consumer, args=(‘B‘,))
p1.start()
c1.start()

1.生产者消费者模型

    通过厨师做包子与顾客吃包子来引出生产者消费者模型，如下图：

    这里，厨师相当于生产者，顾客相当于消费者，顾客吃包子，厨师做包子。做一个假设，如果厨师做包子的速度远远比顾客吃包子的速度要快，有这样一种情况，厨师等顾客吃完一个包子后再做下一个包子（这时我们说厨师与顾客的耦合度高，即厨师做包子与顾客吃包子是紧密相连的），这样显然效率会很低，现实情况中也不可能是这样，因为一旦顾客多时，厨师就有可能忙不过来了。

    可以尝试这样解决：不管顾客有没有吃完包子，厨师也继续做包子，但此时是先把包子放在一个包子柜台中，等顾客有需要时再去包子柜台拿包子。如此一来，厨师和顾客的耦合度就变低了，即厨师做包子并不取决于顾客是否把包子吃完，这样的话效率显然就会高多。

2.Python中的消息队列

    通过上面的例子中，就可以引出消息队列了：包子柜台即相当于Python中的消息队列（当然不只有Python才有消息队列）。根据上面的例子，我们做下面的类比分析。

    类比分析1：厨师和包子相当于是两个线程（假设线程A和线程B），厨师做的包子即相当于是线程A执行后的结果，而线程B的执行需要利用线程A的执行结果，并且，线程A的执行速度比线程B的执行速度要快。

    类比分析2：厨师不会等顾客吃完包子后再做下一个包子，即线程A也不会等线程B使用线程A的执行结果后再去执行下一次功能相同的线程A2，否则程序运行效率会很低。

    类比分析3：厨师把做好的包子放在包子柜台里，顾客吃完一个包子后再去包子柜台取，线程A把执行结果存放在消息队列中，然后再执行下一个功能相同的线程A2，线程B在消息队列中取胜线程A的执行结果，然后再执行下一个功能相同的线程B2，如此类推。

    通过上面的分析，我们就可以知道，通过使用消息队列，我们就可以降低两个线程之间的耦合度，这样就可以提高程序的运行效率。

3.利用消息队列实现Python多线程异步交互

    上面的例子，线程A和线程B的执行速度是不一样的（异步），但却线程B却需要使用线程A的执行结果（交互），通过使用Python的消息队列，就可以实现线程的异步交互。

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


# !/usr/bin/env python


import threading, time
import queue  # 导入消息队列模块
import random  # 导入随机数模块，是为了模拟生产者与消费者速度不一致的情形

q = queue.Queue()  # 实例化一个对象


def Producer(name):  # 生产者函数
    for i in range(20):
        q.put(i)  # 将结果放入消息队列中
        print(‘\033[32;1mProducer %s has made %s baozi....\033[0m‘ % (name, i))
        time.sleep(random.randrange(3))  # 生产者的生产速度，3s内


def Consumer(name):  # 消费者函数
    count = 0
    while count < 20:
        data = http://www.mamicode.com/q.get()  # 取用消息队列中存放的结果>

 利用这个程序，很好地模拟了前面“厨师做包子顾客吃包子”的例子，而从程序的执行结果中也可以看出，线程的执行是异步的，尽管如此，数据还是进行了交互，作用是：在多线程和多线程之间进行数据交互的时候，不会出现数据的阻塞。

    基于前面3点，对Python多线程异步交互应该是有一个比较好的理解了。

4.再谈耦合度的问题

    继续前面“厨师做包子与顾客吃包子”的问题，有一个问题，如果包子的制作是两个厨师完成，即一个做馅，一个包包子，那么正常情况下，两个人的工作需要串行完成才能做完一个包子，即这两个人的耦合度非常高，也就是说他们的工作联系紧密，在这种情况下，我们应该尽可能地减少这种耦合度，这时，只需要在两个厨师之间再加一个队列，即厨师A把馅做好后就放在队列中，厨师B根据自己做包子的快慢去取馅，这样就不会出现厨师A非得等厨师B把一个包子包完了再去做另一个包子的馅，厨师B也不一定需要先等厨师A把手头上的馅做好才去包包子，因为这时队列中已经有馅了，如此因为，再根据两个两项工作速度的不同，给厨师A或厨师B再多增加一名或多名助手，就可以大大地增加整一个做包子的速度了。

    运用在写程序的过程中，也应该要这样去思考。

多进程multiprocessing

要让Python程序实现多进程（multiprocessing），我们先了解操作系统的相关知识。

Unix/Linux操作系统提供了一个fork()系统调用，它非常特殊。普通的函数调用，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。

子进程永远返回0，而父进程返回子进程的ID。这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。

Python的os模块封装了常见的系统调用，其中就包括fork，可以在Python程序中轻松创建子进程：

import os

print(‘Process (%s) start...‘ % os.getpid())
# Only works on Unix/Linux/Mac:
pid = os.fork()
if pid == 0:
    print(‘I am child process (%s) and my parent is %s.‘ % (os.getpid(), os.getppid()))
else:
    print(‘I (%s) just created a child process (%s).‘ % (os.getpid(), pid))

 结果：

Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.

由于Windows没有fork调用，上面的代码在Windows上无法运行。由于Mac系统是基于BSD（Unix的一种）内核，所以，在Mac下运行是没有问题的，推荐大家用Mac学Python！

有了fork调用，一个进程在接到新任务时就可以复制出一个子进程来处理新任务，常见的Apache服务器就是由父进程监听端口，每当有新的http请求时，就fork出子进程来处理新的http请求。

multiprocessing

如果你打算编写多进程的服务程序，Unix/Linux无疑是正确的选择。由于Windows没有fork调用，难道在Windows上无法用Python编写多进程的程序？

由于Python是跨平台的，自然也应该提供一个跨平台的多进程支持。multiprocessing模块就是跨平台版本的多进程模块。

multiprocessing模块提供了一个Process类来代表一个进程对象，下面的例子演示了启动一个子进程并等待其结束：

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
    print(‘Run child process %s (%s)...‘ % (name, os.getpid()))

if __name__==‘__main__‘:
    print(‘Parent process %s.‘ % os.getpid())
    p = Process(target=run_proc, args=(‘test‘,))
    print(‘Child process will start.‘)
    p.start()
    p.join()
    print(‘Child process end.‘)

执行结果如下：

Parent process 928.
Process will start.
Run child process test (929)...
Process end.

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，这样创建进程比fork()还要简单。

join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。

Pool

如果要启动大量的子进程，可以用进程池的方式批量创建子进程：

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print(‘Run task %s (%s)...‘ % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print(‘Task %s runs %0.2f seconds.‘ % (name, (end - start)))

if __name__==‘__main__‘:
    print(‘Parent process %s.‘ % os.getpid())
    p = Pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print(‘Waiting for all subprocesses done...‘)
    p.close()
    p.join()
    print(‘All subprocesses done.‘)

执行结果如下：

Parent process 669.
Waiting for all subprocesses done...
Run task 0 (671)...
Run task 1 (672)...
Run task 2 (673)...
Run task 3 (674)...
Task 2 runs 0.14 seconds.
Run task 4 (673)...
Task 1 runs 0.27 seconds.
Task 3 runs 0.86 seconds.
Task 0 runs 1.41 seconds.
Task 4 runs 1.91 seconds.
All subprocesses done.

代码解读：

对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了。

请注意输出的结果，task 0，1，2，3是立刻执行的，而task 4要等待前面某个task完成后才执行，这是因为Pool的默认大小在我的电脑上是4，因此，最多同时执行4个进程。这是Pool有意设计的限制，并不是操作系统的限制。如果改成：

p = Pool(5)

就可以同时跑5个进程。

由于Pool的默认大小是CPU的核数，如果你不幸拥有8核CPU，你要提交至少9个子进程才能看到上面的等待效果。

子进程

很多时候，子进程并不是自身，而是一个外部进程。我们创建了子进程后，还需要控制子进程的输入和输出。

subprocess模块可以让我们非常方便地启动一个子进程，然后控制其输入和输出。

下面的例子演示了如何在Python代码中运行命令nslookup www.python.org，这和命令行直接运行的效果是一样的：

import subprocess

print(‘$ nslookup www.python.org‘)
r = subprocess.call([‘nslookup‘, ‘www.python.org‘])
print(‘Exit code:‘, r)

运行结果：

$ nslookup www.python.org
Server:        192.168.19.4
Address:    192.168.19.4#53

Non-authoritative answer:
www.python.org    canonical name = python.map.fastly.net.
Name:    python.map.fastly.net
Address: 199.27.79.223

Exit code: 0

如果子进程还需要输入，则可以通过communicate()方法输入：

import subprocess

print(‘$ nslookup‘)
p = subprocess.Popen([‘nslookup‘], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b‘set q=mx\npython.org\nexit\n‘)
print(output.decode(‘utf-8‘))
print(‘Exit code:‘, p.returncode)

上面的代码相当于在命令行执行命令nslookup，然后手动输入：

set q=mx
python.org
exit

运行结果如下：

$ nslookup
Server:        192.168.19.4
Address:    192.168.19.4#53

Non-authoritative answer:
python.org    mail exchanger = 50 mail.python.org.

Authoritative answers can be found from:
mail.python.org    internet address = 82.94.164.166
mail.python.org    has AAAA address 2001:888:2000:d::a6


Exit code: 0

进程间通信

Process之间肯定是需要通信的，操作系统提供了很多机制来实现进程间的通信。Python的multiprocessing模块包装了底层的机制，提供了Queue、Pipes等多种方式来交换数据。

我们以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：

from multiprocessing import Process, Queue
import os, time, random

# 写数据进程执行的代码:
def write(q):
    print(‘Process to write: %s‘ % os.getpid())
    for value in [‘A‘, ‘B‘, ‘C‘]:
        print(‘Put %s to queue...‘ % value)
        q.put(value)
        time.sleep(random.random())

# 读数据进程执行的代码:
def read(q):
    print(‘Process to read: %s‘ % os.getpid())
    while True:
        value = http://www.mamicode.com/q.get(True)
        print(‘Get %s from queue.‘ % value)

if __name__==‘__main__‘:
    # 父进程创建Queue，并传给各个子进程：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 启动子进程pw，写入:
    pw.start()
    # 启动子进程pr，读取:
    pr.start()
    # 等待pw结束:
    pw.join()
    # pr进程里是死循环，无法等待其结束，只能强行终止:
    pr.terminate()

运行结果如下：

Process to write: 50563
Put A to queue...
Process to read: 50564
Get A from queue.
Put B to queue...
Get B from queue.
Put C to queue...
Get C from queue.

在Unix/Linux下，multiprocessing模块封装了fork()调用，使我们不需要关注fork()的细节。由于Windows没有fork调用，因此，multiprocessing需要“模拟”出fork的效果，父进程所有Python对象都必须通过pickle序列化再传到子进程去，所有，如果multiprocessing在Windows下调用失败了，要先考虑是不是pickle失败了。

小结

在Unix/Linux下，可以使用fork()调用实现多进程。

要实现跨平台的多进程，可以使用multiprocessing模块。

进程间通信是通过Queue、Pipes等实现的。

多线程

我们前面提到了进程是由若干线程组成的，一个进程至少有一个线程。

由于线程是操作系统直接支持的执行单元，因此，高级语言通常都内置多线程的支持，Python也不例外，并且，Python的线程是真正的Posix Thread，而不是模拟出来的线程。

Python的标准库提供了两个模块：_thread和threading，_thread是低级模块，threading是高级模块，对_thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。

启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()开始执行：

import time, threading

# 新线程执行的代码:
def loop():
    print(‘thread %s is running...‘ % threading.current_thread().name)
    n = 0
    while n < 5:
        n = n + 1
        print(‘thread %s >>> %s‘ % (threading.current_thread().name, n))
        time.sleep(1)
    print(‘thread %s ended.‘ % threading.current_thread().name)

print(‘thread %s is running...‘ % threading.current_thread().name)
t = threading.Thread(target=loop, name=‘LoopThread‘)
t.start()
t.join()
print(‘thread %s ended.‘ % threading.current_thread().name)

执行结果如下：

thread MainThread is running...
thread LoopThread is running...
thread LoopThread >>> 1
thread LoopThread >>> 2
thread LoopThread >>> 3
thread LoopThread >>> 4
thread LoopThread >>> 5
thread LoopThread ended.
thread MainThread ended.

由于任何进程默认就会启动一个线程，我们把该线程称为主线程，主线程又可以启动新的线程，Python的threading模块有个current_thread()函数，它永远返回当前线程的实例。主线程实例的名字叫MainThread，子线程的名字在创建时指定，我们用LoopThread命名子线程。名字仅仅在打印时用来显示，完全没有其他意义，如果不起名字Python就自动给线程命名为Thread-1，Thread-2……

Lock

多线程和多进程最大的不同在于，多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响，而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。

来看看多个线程同时操作一个变量怎么把内容给改乱了：

import time, threading

# 假定这是你的银行存款:
balance = 0

def change_it(n):
    # 先存后取，结果应该为0:
    global balance
    balance = balance + n
    balance = balance - n

def run_thread(n):
    for i in range(100000):
        change_it(n)

t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)

我们定义了一个共享变量balance，初始值为0，并且启动两个线程，先存后取，理论上结果应该为0，但是，由于线程的调度是由操作系统决定的，当t1、t2交替执行时，只要循环次数足够多，balance的结果就不一定是0了。

原因是因为高级语言的一条语句在CPU执行时是若干条语句，即使一个简单的计算：

balance = balance + n

也分两步：

1. 计算balance + n，存入临时变量中；
2. 将临时变量的值赋给balance。

也就是可以看成：

x = balance + n
balance = x

由于x是局部变量，两个线程各自都有自己的x，当代码正常执行时：

初始值 balance = 0

t1: x1 = balance + 5 # x1 = 0 + 5 = 5
t1: balance = x1     # balance = 5
t1: x1 = balance - 5 # x1 = 5 - 5 = 0
t1: balance = x1     # balance = 0

t2: x2 = balance + 8 # x2 = 0 + 8 = 8
t2: balance = x2     # balance = 8
t2: x2 = balance - 8 # x2 = 8 - 8 = 0
t2: balance = x2     # balance = 0

结果 balance = 0

但是t1和t2是交替运行的，如果操作系统以下面的顺序执行t1、t2：

初始值 balance = 0

t1: x1 = balance + 5  # x1 = 0 + 5 = 5

t2: x2 = balance + 8  # x2 = 0 + 8 = 8
t2: balance = x2      # balance = 8

t1: balance = x1      # balance = 5
t1: x1 = balance - 5  # x1 = 5 - 5 = 0
t1: balance = x1      # balance = 0

t2: x2 = balance - 8  # x2 = 0 - 8 = -8
t2: balance = x2   # balance = -8

结果 balance = -8

究其原因，是因为修改balance需要多条语句，而执行这几条语句时，线程可能中断，从而导致多个线程把同一个对象的内容改乱了。

两个线程同时一存一取，就可能导致余额不对，你肯定不希望你的银行存款莫名其妙地变成了负数，所以，我们必须确保一个线程在修改balance的时候，别的线程一定不能改。

如果我们要确保balance计算正确，就要给change_it()上一把锁，当某个线程开始执行change_it()时，我们说，该线程因为获得了锁，因此其他线程不能同时执行change_it()，只能等待，直到锁被释放后，获得该锁以后才能改。由于锁只有一个，无论多少线程，同一时刻最多只有一个线程持有该锁，所以，不会造成修改的冲突。创建一个锁就是通过threading.Lock()来实现：

balance = 0
lock = threading.Lock()

def run_thread(n):
    for i in range(100000):
        # 先要获取锁:
        lock.acquire()
        try:
            # 放心地改吧:
            change_it(n)
        finally:
            # 改完了一定要释放锁:
            lock.release()

当多个线程同时执行lock.acquire()时，只有一个线程能成功地获取锁，然后继续执行代码，其他线程就继续等待直到获得锁为止。

获得锁的线程用完后一定要释放锁，否则那些苦苦等待锁的线程将永远等待下去，成为死线程。所以我们用try...finally来确保锁一定会被释放。

锁的好处就是确保了某段关键代码只能由一个线程从头到尾完整地执行，坏处当然也很多，首先是阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率就大大地下降了。其次，由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，可能会造成死锁，导致多个线程全部挂起，既不能执行，也无法结束，只能靠操作系统强制终止。

多核CPU

如果你不幸拥有一个多核CPU，你肯定在想，多核应该可以同时执行多个线程。

如果写一个死循环的话，会出现什么情况呢？

打开Mac OS X的Activity Monitor，或者Windows的Task Manager，都可以监控某个进程的CPU使用率。

我们可以监控到一个死循环线程会100%占用一个CPU。

如果有两个死循环线程，在多核CPU中，可以监控到会占用200%的CPU，也就是占用两个CPU核心。

要想把N核CPU的核心全部跑满，就必须启动N个死循环线程。

试试用Python写个死循环：

import threading, multiprocessing

def loop():
    x = 0
    while True:
        x = x ^ 1

for i in range(multiprocessing.cpu_count()):
    t = threading.Thread(target=loop)
    t.start()

启动与CPU核心数量相同的N个线程，在4核CPU上可以监控到CPU占用率仅有102%，也就是仅使用了一核。

但是用C、C++或Java来改写相同的死循环，直接可以把全部核心跑满，4核就跑到400%，8核就跑到800%，为什么Python不行呢？

因为Python的线程虽然是真正的线程，但解释器执行代码时，有一个GIL锁：Global Interpreter Lock，任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都给上了锁，所以，多线程在Python中只能交替执行，即使100个线程跑在100核CPU上，也只能用到1个核。

GIL是Python解释器设计的历史遗留问题，通常我们用的解释器是官方实现的CPython，要真正利用多核，除非重写一个不带GIL的解释器。

所以，在Python中，可以使用多线程，但不要指望能有效利用多核。如果一定要通过多线程利用多核，那只能通过C扩展来实现，不过这样就失去了Python简单易用的特点。

不过，也不用过于担心，Python虽然不能利用多线程实现多核任务，但可以通过多进程实现多核任务。多个Python进程有各自独立的GIL锁，互不影响。

小结

多线程编程，模型复杂，容易发生冲突，必须用锁加以隔离，同时，又要小心死锁的发生。

Python解释器由于设计时有GIL全局锁，导致了多线程无法利用多核。多线程的并发在Python中就是一个美丽的梦。

进程池　　

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

• apply
• apply_async

from  multiprocessing import Process,Pool
import time
 
def Foo(i):
    time.sleep(2)
    return i+100
 
def Bar(arg):
    print(‘-->exec done:‘,arg)
 
pool = Pool(5)
 
for i in range(10):
    pool.apply_async(func=Foo, args=(i,),callback=Bar)
    #pool.apply(func=Foo, args=(i,))
 
print(‘end‘)
pool.close()
pool.join()#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。