python线程问题汇总（日常整理）

以下问题来自 本小白的一些不懂的地方。很多地方借于百度和一些我的理解。特此整理

线程是什么

线程是进程中执行运算的最小单位，是进程中的一个实体，是被系统独立调度和分派的基本单位，线程自己不拥有系统资源，只拥有一点在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程，同一进程中的多个线程之间可以并发执行。

一个进程下的子线程与主线程共享同一片数据空间

进程是什么

正在进行的一个过程或者说一个任务，而负责执行任务的则是CPU，进程本身是

一个抽象的概念,即进程就是一个过程、一个任务。

CPU描述的是一个程序的执行过程.

说人话线程和进程是什么

线程是系统调度的基本单位

进程是系统正在运行的任务

显而易见的 ：N个线程组成了一个进程

线程和进程的关系

线程是属于进程的，线程运行在进程空间内，同一进程所产生的线程共享同一内存空间，当进程退出时该进程所产生的线程都会被强制退出并清除。线程可与属于同一进程的其它线程共享进程所拥有的全部资源，但是其本身基本上不拥有系统资源，只拥有一点在运行中必不可少的信息(如程序计数器、一组寄存器和栈)。

简而言之

一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程。

资源分配给进程，同一进程的所有线程共享该进程的所有资源。

处理机分给线程，即真正在处理机上运行的是线程

线程在执行过程中，需要协作同步。不同进程的线程间要利用消息通信的办法实现同步。线程是指进程内的一个执行单元,也是进程内的可调度实体.

线程与进程的区别

调度：线程作为调度和分配的基本单位，进程作为拥有资源的基本单位

并发性：不仅进程之间可以并发执行，同一个进程的多个线程之间也可并发执行

拥有资源：进程是拥有资源的一个独立单位，线程不拥有系统资源，但可以访问隶属于进程的资源.

系统开销：在创建或撤消进程时，由于系统都要为之分配和回收资源，导致系统的开销明显大于创建或撤消线程时的开销。

多线程

上面说的都是多线程。什么是多线程？就是开启多个线程一起工作。多线程有什么好处？

使用线程可以把占据长时间的程序中的任务放到后台去处理。

用户界面可以更加吸引人，这样比如用户点击了一个按钮去触发某些事件的处理，可以弹出一个进度条来显示处理的进度

程序的运行速度可能加快

在一些等待的任务实现上如用户输入、文件读写和网络收发数据等，线程就比较有用了。在这种情况下我们可以释放一些珍贵的资源如内存占用等等。

线程的创建

import threading
import time

def run(n):
print('----task', n)
time.sleep(2)

#t_obg = []
for i in range(50):
t = threading.Thread(target=run, args=('t-%s' %i, ))
#t.setDaemon(True)
t.start()
t_obg.append(t)

#for j in t_obg:
#    j.join()
print('over')

# 两种方法 的结果一致，一种是类，一种是方法
import threading
import time

class Mythread(threading.Thread):
def __init__(self, n):
super(Mythread,self).__init__()
self.n = n

def run(self):
print('----task', self.n)
time.sleep(2)

if __name__ == '__main__':

# t_obg = []
for i in range(50):
t = Mythread('t-%s' %i)
#t.setDaemon(True)
t.start()
#     t_obg.append(t)
#
# for j in t_obg:
#    j.join()
print('over')

在python3中线程线程创建十分简单,如上所见。首先导入了一个threading包和一个time包（为什么使用threading而不是thread，首先这是python3不是python2.x 那种落后的东西就放弃吧，说实话我学的时候就看的threading，thread没看过）

定义了一个run方法，随便打印了一些字 并睡眠2s

重点来了：

threading.Thread(target=run, args=(‘t-%s’ %i, ))创建一个线程，target传递的是函数，args传递的是target函数需要的参数元组的形式传递

setDaemon是设置守护线程

.start()就是开启线程

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

就当作这里：！！！单纯的创建只需要threading.Thread（）.start（）

（感觉有坑 日后在补）

避免使用thread模块

推荐使用更高级别的threading模块，而不是用thread模块有很多原因。threading模块更加先进，有更好的线程支持，并且thread模块中的一些属性会和threading模块有冲突，另一个原因是低级别的thread模块拥有的同步原语很少（其实只有一个），而threading模块则有很多。

避免使用thread模块的另一个原因是它对于进程何时退出没有控制。当主线程结束时，所有其他线程也都强制结束，不会发出警告或者进行适当的清理。如上所述，至少threading模块能够确保重要的子线程在进程退出前结束（join）

只建议哪些想访问线程的更底层级别的专家使用thread模块。为了增强这点，在python3中改模块被重命名为_thread。

守护线程

如果你设置一个线程为守护线程，，就表示你在说这个线程是不重要的，在进程退出的时候，不用等待这个线程退出。

如果你的主线程在退出的时候，不用等待那些子线程完成，那就设置这些线程的daemon属性。即，在线程开始（thread.start()）之前，调用setDaemon（）函数，设定线程的daemon标志。（thread.setDaemon(True)）就表示这个线程“不重要”。

如果你想等待子线程完成再退出，那就什么都不用做。，或者显示地调用thread.setDaemon(False)，设置daemon的值为false。新的子线程会继承父线程的daemon标志。整个Python会在所有的非守护线程退出后才会结束，即进程中没有非守护线程存在的时候才结束。

setDaemon()需要在start之前写

没有设置守护线程

import threading
import time

def fuc():
print('【】----start',)
time.sleep(2)
print('【】----end', )

if __name__ == '__main__':
t = threading.Thread(target=fuc,args=())
t.start()
time.sleep(1)
print('main thread end')

没有设置守护线程（默认setDeamon(False)）

19773
运行结果如下：

【】----start
main thread end
【】----end

Process finished with exit code 0

结论：程序等待子线程结束才完成

设置守护线程

import threading
import time

def fuc():
print('【】----start',)
time.sleep(2)
print('【】----end', )

if __name__ == '__main__':
t = threading.Thread(target=fuc,args=())
t.setDaemon(True)
t.start()
time.sleep(1)
print('main thread end')

运行结果如下：

【】----start
main thread end

Process finished with exit code 0

结论：程序没有等待子线程完成就结束了

简单说就是 如果设置了守护线程，就像收了一个仆从，当皇帝死了以后，不需要管仆从是否老死，都要陪葬Twt

补充一点很重要的是 主线程也是线程，子线程是在主线程下开辟的

join()

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

python 默认参数创建线程后，不管主线程是否执行完毕，都会等待子线程执行完毕才一起退出，有无join结果一样

如果创建线程，并且设置了daemon为true，及threading.setDaemon(True),则主线程执行完毕后自动退出，不会等待子线程的执行结果。而且随着主线程的退出，子线程也消亡

join方法的作用是阻塞，等待子线程结束，join方法有一个参数是timeout，即如果主线程等待timeout，子线程还没有结束，则主线程强制结束子线程。

如果线程daemon属性为False，则join里的timeout参数无效，主线程会一直等待子线程结束。

如果线程daemon属性为True，则join里的timeout参数是有效的，主线程会等待timeout时间后，结束子线程。（此处有坑）即如果同时又N个子线程join（timeout），那么实际上主线程会等待的超时时间最长为N*timeout,因为每个子线程的超时开始时刻是上一个子线程的超时结束的时刻。

import threading, time
def doThreadTest():
print 'start thread time:', time.strftime('%H:%M:%S')
time.sleep(10)
print 'stop thread time:', time.strftime('%H:%M:%S')

threads = []
for i in range(3):
thread1 = threading.Thread(target = doThreadTest)
thread1.setDaemon(True)

threads.append(thread1)

for t in threads:
t.start()

for t in threads:
t.join(1)
print 'stop main thread'

也就是说join是为了保证主线程在子线程结束后在结束，保证线程的安全

同步原语

锁

GIL

全局解释器锁（GIL），Python代码的执行是有Python虚拟机进行控制的。python在设计时这样考虑，在主循环中同时只能有一个控制线程在执行，就想单核CPU系统中的多进程一样。内存中可以有许多程序，在是在任意给定时刻只能有一个程序在运行，同理。尽管Python解释器中可以运行多个线程，但是在任意给定时刻只有一个线程会被解释器执行。

对python虚拟机的访问是由全局解释器锁（GIL）控制的。这个锁就是用来保证同时只能有一个线程运行的。在多线程环境中，Python虚拟机将按照下面所述的方式执行：

设置GIL

切换进一个线程中去运行

执行下面的操作之一：

a. 指定数量的字节码指令

b.线程主动让出控制权（可以调用time.sleep(0)来完成）

把线程设置会睡眠状态（切换出线程）

解锁GIL

重复上述步骤

就是说：无论你启多少个线程，你有多少个cpu, Python在执行的时候会淡定的在同一时刻只允许一个线程运行。

由于GIL的存在 python代码属于执行CPU密集类型，多个线程代码很有可能是线性执行。这种情况下多线程是鸡肋，效率可能还不如单线程因为有context switch

但是：如果代码是IO密集型 ，多线程可以明显提高效率

既然已经有GIL了为什么还需要锁？

为了保证在某一部分数据读取时串行进行，GIL只是保证了在同一时间只有一个线程在进行操作，并不能保证在同一时间需要的数据被只被一个线程所持有。简单说就是为了保证线程的数据安全

GIL：全局解释器锁

作用：保证同一时刻，只有一个线程被CPU执行，无论你有多少个线程。

为什么这里还需要lock? 注意啦，这里的lock是用户级的lock,跟那个GIL没关系 ，具体我们通过下图进行讲解



　　既然用户程序已经自己有锁了，那为什么C python还需要GIL呢？加入GIL主要的原因是为了降低程序的开发的复杂度，比如现在的你写python不需要关心内存回收的问题，因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，这可以说是Python早期版本的遗留问题。

加锁

语句	解释
thread.allocate_lock()	返回一个新Lock对象，即为一个新锁
lock.acquire()	相当于P操作，得到一个锁
lock.release()	相当于V操作，释放一个锁
lock.locked()	已经得到了锁,返回True(一般用不到)

互斥锁

（由于并发状态下，操作系统对多个进程进行调度，而多个进程可能都有操作硬件的需求，这时就会产生多个进程对资源的共享，而共享意味着竞争，会产生许多问题。这样就需要一种机制或者手段去解决竞争，使竞争变得有序化，从而使共享资源按照预定的结果去获取。这种手段就是加互斥锁）

import threading
import os
import time

class demo():
def __init__(self,d):
self.dic = d

def check_ticket(self):
print('剩余【%s】票' % self.dic['ticket'])
def deal_ticket(self):
if self.dic['ticket'] > 0:
self.dic['ticket'] -= 1
print('%s购票成功' % os.getpid())
else:
print('%s购票失败' % os.getpid())
def buy_ticket(self,):
time.sleep(2)
locks.acquire()
self.check_ticket()
self.deal_ticket()
locks.release()

if __name__ == '__main__':
locks = threading.Lock()
d = demo({'ticket': 10})
for i in range(10):
t = threading.Thread(target=d.buy_ticket,)
t.start()

打印结果:

剩余【10】票
7816购票成功
剩余【9】票
7816购票成功
剩余【8】票
7816购票成功
剩余【7】票
7816购票成功
剩余【6】票
7816购票成功
剩余【5】票
7816购票成功
剩余【4】票
7816购票成功
剩余【3】票
7816购票成功
剩余【2】票
7816购票成功
剩余【1】票
7816购票成功

Process finished with exit code 0

稍作修改把锁去掉

import threading
import os
import time

class demo():
def __init__(self,d):
self.dic = d

def check_ticket(self):
print('剩余【%s】票' % self.dic['ticket'])
def deal_ticket(self):
if self.dic['ticket'] > 0:
self.dic['ticket'] -= 1
print('%s购票成功' % os.getpid())
else:
print('%s购票失败' % os.getpid())
def buy_ticket(self,):
time.sleep(2)
#locks.acquire()
self.check_ticket()
self.deal_ticket()
#locks.release()

if __name__ == '__main__':
#locks = threading.Lock()
d = demo({'ticket': 10})
for i in range(10):
t = threading.Thread(target=d.buy_ticket,)
t.start()

打印结果：

剩余【10】票
2764购票成功
剩余【9】票
2764购票成功
剩余【8】票
剩余【8】票
剩余【8】票
2764购票成功
剩余【7】票
2764购票成功
剩余【6】票
2764购票成功
2764购票成功
2764购票成功
剩余【3】票
2764购票成功
剩余【2】票
剩余【2】票
2764购票成功
2764购票成功

Process finished with exit code 0

死锁

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源都

简单的说死锁是由于一个资源被多次调用，而多次调用方都未能释放该资源就会造成死锁。

递归锁

import threading
import time

def run1():
print('grrab the first part data')
lock.acquire()
global num
num += 1
lock.release()
return num

def run2():
print('grrab 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)

num, num2 = 0, 0
lock = threading.Lock()
for i in range(10):
t = threading.Thread(target=run3)
t.start()

while threading.active_count() != 1:
print(threading.active_count())
else:
print('---all thread done---')
print(num, num2)

这个的打印结果：

11

11

11

……

成功造成了死锁

（至于为什么打印11，是因为print(threading.active_count())，这句话是在打印当前的线程数，我们开启了10个子线程和一个主线程，so 打印11）

至于解决方法：

很简单，在python3中除了提供了Lock（）方法还提供了RLock（）

import threading

def run1():
print('grrab the first part data')
lock.acquire()
global num
num += 1
lock.release()
return num

def run2():
print('grrab 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)

num, num2 = 0, 0
# lock = threading.Lock()
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 thread done---')
print(num, num2)

RLock就是用来解决由多个锁嵌套造成死锁问题

除了上面的RLock解决死锁的方法：

引入看门狗计数器。当现成正常运行的时候，每隔一段时间重置计数器，在没有死锁的情况下正常进行。一旦发生了死锁，由于无法重置计数器导致定时器超时，这是程序会通过重启自身恢复到正常状态

避免死锁。emm。避免死锁的主要思想是，单纯地按照对象id递增的顺序加锁不会产生循环依赖，而循环依赖是 死锁的一个必要条件，从而避免程序进入死锁状态。

信号量

信号量semaphore 是一个变量，控制着对公共资源或者临界区的访问。信号量维护着一个计数器，指定可同时访问资源或者进入临界区的线程数。

线程中，信号量主要是用来维持有限的资源，使得在一定时间使用该资源的线程只有指定的数量

'''
每次有一个线程获得信号量时，计数器-1。若计数器为0，其他线程就停止访问信号量，直到另一个线程释放信号量。
'''

import threading, time

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

if __name__ == '__main__':
semaphore = threading.BoundedSemaphore(5)#最多允许5个线程同时进行

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

while threading.active_count() != 1:
# print(threading.active_count())
pass
else:
print('over')

Event

用threading.Event():可以实现线程之间的通讯，使一个线程阻塞，执行另一个线程

import time
import threading

event = threading.Event()
event.set()

#模拟红绿灯
def lighter():
count = 0
while True:
if count > 20 and count <30:
event.clear()
print('\033[41;1m red light is on ...\033[0m')
elif count > 30:
count = 0
event.set()
else:
print('\033[46;1m green light is on ...\033[0m')
time.sleep(1)
count += 1

#模拟车辆
def car(name):
while True:
if event.is_set(): # 判断是否阻塞 ，如果不阻塞
print('[%s] running..' % name)
time.sleep(1)
else:
print('[%s] sees red light ,waitting' % name)
event.wait()

light = threading.Thread(target=lighter, )
car = threading.Thread(target=car, args=('car1',))

light.start()
car.start()

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

在全局定义一个Flag，当Flag为False，那么当程序执行even.wait时 方法就会阻塞，当Flag为True,even.wait就不会阻塞

clear 将Flag设置为Flase

set 将Flag设置为True

队列

在python中的队列是线程中最常用的交换数据的形式。queue模块是提供队列操作的模块.

queue模块就是提供线程通讯的机制，从而让线程之间可以相互分享数据。具体而言，就是创建一个队列，让生产者（线程）在其中放入新的商品，而消费者（线程）进行消费

属性	描述
queue模块的类
Queue(maxsize=0)	创建一个先入先出队列。如果给定最大值，则在队列没有空间时阻塞；否则（没有给定最大值），为无限队列
LifoQueue(maxsize=0)	创建一个先入后出队列。如果给定最大值，则在队列没有空间时阻塞；否则（没有给定最大值），为无限队列
PriorityQueue(maxsize=0)	创建一个优先级队列。如果给定最大值，则在队列没有空间时阻塞；否则（没有给定最大值），为无限队列
queue异常
Empty	当对空队列调用get*（）方法时抛出的异常
Full	当对已满队列调用put*（）方法时抛出的异常
queue对象方法
qsize（）	返回队列大小（由于返回队列大小可能被其他线程修改，所以该值为近似值）
empty（）	如果队列为空，则返回True；否则，返回False
full（）	如果队列已满，则返回True；否则,返回False
put(item,block=True,timeout=None)	将item放入队列。如果block为True（默认）且timeout为None，则在有可用空间之前阻塞；如果timeout为正值，则最多阻塞timeout秒，如果block为False。则抛出Empty异常
put_nowait(item)	和put(item,False)相同
get(block=True,timeout=None)	从队列中获取元素，如果指定block（非零），则一直阻塞到有可用的元素为止
get_nowait（）	和get（False）相同
task_done()	用于表示队列中的某个元素以执行完成，该方法会被下面的join（）使用
join()	在队列中所有元素执行完毕并调用上面的task_done()新号之前，保持阻塞

基本FIFO队列

FIFO即First in First Out,先进先出。Queue提供了一个基本的FIFO容器，使用方法很简单,maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。

import queue

q = queue.Queue(maxsize=3)  # 先入先出 maxsize 设置队列最大长度

q.put(1,block=False)    # 存入数据
q.put(2)
print(q.get())   # 获取数据
print(q.get())
print(q.get(timeout=1))
print(q.qsize())
print(q.full())
print(q.get_nowait())  # 获取数据 当没有数据时直接抛出异常
print(q.get_nowait())

基本的lIFO队列

LIFO即Last in First Out,后进先出。与栈的类似，使用也很简单,maxsize用法同上

import queue
q = queue.LifoQueue()  # 后进先出

q.put(1)
q.put(2)
q.put(3)

print(q.get())
print(q.get())
print(q.get())

优先级队列

import queue
q = queue.PriorityQueue()  # 存储时设置优先级

q.put((2, 'asq2'))
q.put((1, 'asw1'))
q.put((4, 'ase3'))
q.put((10, 'asr4'))

print(q.get())
print(q.get())
print(q.get())
print(q.get())

简单的应用：多线程中取值

import queue
import threading
import time
import random

q=queue.Queue(0)
NUM_WORKERS = 3

class MyThread(threading.Thread):
"""A worker thread."""
def __init__(self, input, worktype):
self._jobq = input
self._work_type = worktype
threading.Thread.__init__(self)
def run(self):
"""
Get a job and process it.
Stop when there's no more jobs
"""
while True:
if self._jobq.qsize()>0:
job = self._jobq.get()
worktype=self._work_type
self._process_job(job,worktype)
else:
break
def _process_job(self, job,worktype):
"""
Do useful work here.
worktype: let this thread do different work
1,do list
2,do item
3,,,
"""
doJob(job)

def doJob(job):
"""
do work function 1
"""
time.sleep(random.random()*3)
print("doing ",job)
if __name__=='__main__':

print("begin...")
#put some work to q
for i in range(NUM_WORKERS*2):
q.put(i)
#print total job q's size
print("job q'size",q.qsize())
#start threads to work
for x in range(NUM_WORKERS):
MyThread(q,x).start()
#if q is not empty, wait
#while q.qsize()>0:
#    time.sleep(0.1)

双向队列

q = queue.deque()
q.append(123)
q.append(456)
q.appendleft(780)
print(q.pop())
print(q.popleft())

生产者消费者模型

解决什么问题,使用场景

从下面图中可以发现生产者和消费者之间用中间类似一个队列一样的东西串起来。这个队列可以想像成一个存放产品的“仓库”，生产者只需要关心这个“仓库”，并不需要关心具体的消费者，对于生产者而言甚至都不知道有这些消费者存在。对于消费者而言他也不需要关心具体的生产者，到底有多少生产者也不是他关心的事情，他只要关心这个“仓库”中还有没有东西。这种模型是一种松耦合模型。这样可以回答我上面提出的第一个问题。这个模型的产生就是为了复用和解耦。比如常见的消息框架（非常经典的一种生产者消费者模型的使用场景）ActiveMQ。发送端和接收端用Topic进行关联。这个Topic可以理解为我们这里“仓库”的地址，这样就可以实现点对点和广播两种方式进行消息的分发。



简单的demo：

import threading
import time
import queue

def Producer(name):
'''生产者'''
count = 1
while True:
print('[%s] 生产了 %s 个骨头'%(name, count))
q.put('骨头 %s' %count)
count += 1
# time.sleep(1)

def Consumer(name):
'''消费者'''
while True:
print('[%s]吃了[%s]'%(name,q.get()))
# time.sleep(1)

import threading
import time
import queue

class Producer(threading.Thread):
def __init__(self,name):
super(Producer,self).__init__(name = name)

def run(self):
'''生产者'''
count = 1
while True:
print('[%s] 生产了 %s 个骨头'%(self.name, count))
q.put('骨头 %s' %count)
count += 1
time.sleep(1)

class Consumer(threading.Thread):
def __init__(self,name):
super(Consumer,self).__init__(name = name)

def run(self):
'''消费者'''
while True:
print('[%s]吃了[%s]'%(self.name,q.get()))
time.sleep(1.5)

if __name__ == '__main__':
q = queue.Queue(maxsize=10)
p = Producer('Main')
c = Consumer('Job')
p.start()
c.start()

生产者消费者模型的优点：

1、解耦

假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化， 可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。

2、支持并发

由于生产者与消费者是两个独立的并发体，他们之间是用缓冲区作为桥梁连接，生产者只需要往缓冲区里丢数据，就可以继续生产下一个数据，而消费者只需要从缓冲区了拿数据即可，这样就不会因为彼此的处理速度而发生阻塞。

3、支持忙闲不均

缓冲区还有另一个好处。如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。 等生产者的制造速度慢下来，消费者再慢慢处理掉。

并发与并行

并发：就是同时做多件事情。

例如：终端用户程序利用并发功能，在输入数据的同时响应用户输入。服务器利用并发，在处理第一个请求的同时响应第二个请求。只要你希望程序同时做多件事情，就需要并发。

很多人看到“并发”就会想到“多线程”，其实他们是有区别的。多线程只是并发的一种形式，但不是唯一形式

并行：就是把正在执行的大量任务分割成小块，分配给多个同时运行的线程。

一般情况下，为了让CPU充分利用，并行处理都会采用多线程。

所以说：并行处理是多线程的一种，而多线程是并发的一种。

还有一种非常重要但很多人不熟悉的并发类型：异步编程，它也是并发的一种形式。

线程池

在使用多线程处理任务时也不是线程越多越好，由于在切换线程的时候，需要切换上下文环境，依然会造成cpu的大量开销。为解决这个问题，线程池的概念被提出来了。预先创建好一个较为优化的数量的线程，让过来的任务立刻能够使用，就形成了线程池。在python中，没有内置的较好的线程池模块，需要自己实现或使用第三方模块。下面是一个简单的线程池：

import threading,time,os,queue

class ThreadPool(object):
def __init__(self,maxsize):
self.maxsize = maxsize
self._q = queue.Queue(self.maxsize)
for i in range(self.maxsize):
self._q.put(threading.Thread)

def getThread(self):
return self._q.get()

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

def fun(num,p):
print('this is thread [%s]'%num)
time.sleep(1)
p.addThread()

if __name__ == '__main__':
pool = ThreadPool(2)
for i in range(103):
t = pool.getThread()
a = t(target = fun,args = (i,pool))
a.start()

"""
一个基于thread和queue的线程池，以任务为队列元素，动态创建线程，重复利用线程，
通过close和terminate方法关闭线程池。
"""
import queue
import threading
import contextlib
import time

# 创建空对象,用于停止线程
StopEvent = object()

def callback(status, result):
"""
根据需要进行的回调函数，默认不执行。
:param status: action函数的执行状态
:param result: action函数的返回值
:return:
"""
pass

def action(thread_name,arg):
"""
真实的任务定义在这个函数里
:param thread_name: 执行该方法的线程名
:param arg: 该函数需要的参数
:return:
"""
# 模拟该函数执行了0.1秒
time.sleep(0.1)
print("第%s个任务调用了线程 %s，并打印了这条信息！" % (arg+1, thread_name))

class ThreadPool:

def __init__(self, max_num, max_task_num=None):
"""
初始化线程池
:param max_num: 线程池最大线程数量
:param max_task_num: 任务队列长度
"""
# 如果提供了最大任务数的参数，则将队列的最大元素个数设置为这个值。
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 put(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):
"""
创建一个线程
"""
# 每个线程都执行call方法
t = threading.Thread(target=self.call)
t.start()

def call(self):
"""
循环去获取任务函数并执行任务函数。在正常情况下，每个线程都保存生存状态，
直到获取线程终止的flag。
"""
# 获取当前线程的名字
current_thread = threading.currentThread().getName()
# 将当前线程的名字加入已实例化的线程列表中
self.generate_list.append(current_thread)
# 从任务队列中获取一个任务
event = self.q.get()
# 让获取的任务不是终止线程的标识对象时
while event != StopEvent:
# 解析任务中封装的三个参数
func, arguments, callback = event
# 抓取异常，防止线程因为异常退出
try:
# 正常执行任务函数
result = func(current_thread, *arguments)
success = True
except Exception as e:
# 当任务执行过程中弹出异常
result = None
success = False
# 如果有指定的回调函数
if callback is not None:
# 执行回调函数，并抓取异常
try:
callback(success, result)
except Exception as e:
pass
# 当某个线程正常执行完一个任务时，先执行worker_state方法
with self.worker_state(self.free_list, current_thread):
# 如果强制关闭线程的flag开启，则传入一个StopEvent元素
if self.terminal:
event = StopEvent
# 否则获取一个正常的任务，并回调worker_state方法的yield语句
else:
# 从这里开始又是一个正常的任务循环
event = self.q.get()
else:
# 一旦发现任务是个终止线程的标识元素，将线程从已创建线程列表中删除
self.generate_list.remove(current_thread)

def close(self):
"""
执行完所有的任务后，让所有线程都停止的方法
"""
# 设置flag
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)

# 该装饰器用于上下文管理
@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录空闲的线程，或从空闲列表中取出线程处理任务
"""
# 将当前线程，添加到空闲线程列表中
state_list.append(worker_thread)
# 捕获异常
try:
# 在此等待
yield
finally:
# 将线程从空闲列表中移除
state_list.remove(worker_thread)

# 调用方式
if __name__ == '__main__':
# 创建一个最多包含5个线程的线程池
pool = ThreadPool(5)
# 创建100个任务，让线程池进行处理
for i in range(100):
pool.put(action, (i,), callback)
# 等待一定时间，让线程执行任务
time.sleep(3)
print("-" * 50)
print("\033[32;0m任务停止之前线程池中有%s个线程，空闲的线程有%s个！\033[0m"
% (len(pool.generate_list), len(pool.free_list)))
# 正常关闭线程池
pool.close()
print("任务执行完毕，正常退出！")
# 强制关闭线程池
# pool.terminate()
# print("强制停止任务！")

事件驱动模型

http://www.cnblogs.com/nulige/p/6297829.html

协程

http://python.jobbole.com/86481/