并发系列是一个很庞大的知识体系,要想完全弄明白是挺困难的,因为最近打算阅读Tornado源码, 其介绍谈到了内部使用了异步非阻塞的调用方式。之前也没有深入了解,这次就借此好好整理一下。
线程(threading模块)
线程是应用程序运行的最小单元,在同一个进程中,可以并发开启多个线程,每个线程拥有自己的栈(存放临时变量),同时相互之间是共享资源的。
Python中使用threading模块来开启多线程
import threading, time def func(n): time.sleep(2)
print(time.time(),n)
if __name__ == '__main__': for i in range(10): t = threading.Thread(target=func, args=(1,)) t.start() print('主线程结束')
'结果'
主线程结束 1532921321.058243 1 1532921321.058243 1 1532921321.058243 1 1532921321.058243 1 ...
或者通过自定义类继承Thread,并且重写run方法
import threading,time class Mythread(threading.Thread): def __init__(self,name): super().__init__() self.name = name def run(self): time.sleep(1) print('hello %s'%self.name) if __name__ == '__main__': for i in range(5): m = Mythread('py') m.start() print('---主线程结束---')
执行顺序如下
主线程和子线程之间是相互独立的,但是主线程运行完毕会等待子线程的运行,直到完毕,才回收资源。
Thread对象可调用的方法
hread实例对象的方法 # isAlive(): 返回线程是否活动的。 # getName(): 返回线程名。 # setName(): 设置线程名。
#join():使主线程阻塞,直到该线程结束
threading模块提供的一些方法: # threading.currentThread(): 返回当前的线程变量。 # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前,不包括启动前和终止后的线程。 # threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果。
守护线程(setDaemon)
如果一个线程设置为守护线程,那么它将会和主线程一起结束,而主线程会等待所有的非守护线程的子线程结束而退出。因此可以认为,守护线程是“不重要的线程”,主线程不等它。
import threading, time """ 设置两个线程 """ def func1(): print('--非守护线程开始--') time.sleep(2) print('--非守护线程结束--') def func2(): print('--守护线程开始--') time.sleep(4) print('--守护线程结束--') if __name__ == '__main__': t1 = threading.Thread(target=func1,args=()) t2 = threading.Thread(target=func2,args=()) t2.setDaemon(True) t1.start() t2.start() ''' --非守护线程开始-- --守护线程开始-- --非守护线程结束-- 守护线程还没运行完,主线程就结束了 '''
而线程之间共享数据,必然会导致同时操作数据时的混乱,影响数据安全
import threading,time def func(): #开始处理数据 global n a=n+1 time.sleep(0.0001) n =a # 结束处理 if __name__ == '__main__': n=0 li =[] for i in range(1000): t=threading.Thread(target=func,args=()) li.append(t) t.start() for i in li: i.join() #等待子线程全部执行完 print(n) #253 ''' 我们希望能从0加到1000,但是由于有多个线程会拿到数据, 如果处理速度慢,就会使数据混乱 '''
因此,对数据进行加锁就很有必要了。
互斥锁(Lock)
通过获取锁对象,访问共有数据,最后释放锁来完成一次操作,一旦某个线程获取了锁,当这个线程被切换时,下个个进程无法获取该公有数据
import threading,time def func(): #开始处理数据 global n lock.acquire() #获取 a=n+1 time.sleep(0.00001) n =a lock.release() #释放 # 结束处理 if __name__ == '__main__': n=0 lock=threading.Lock() li =[] for i in range(1000): t=threading.Thread(target=func,args=()) li.append(t) t.start() for i in li: i.join() #等待子线程全部执行完 print(n) #1000
通过同步的互斥锁来保证数据安全相比于线程串行运行而言,如每个线程start之前都使用.join()方法,无疑速度更快,因为它就只有在访问数据的时候是串级的,其他的情况下是是并发的(虽然也不能是并行运行,因为GIL,之后会谈到)。
再看如下情况
死锁
import threading if __name__ == '__main__': lock=threading.Lock() lock.acquire() lock.acquire() #程序卡住,只能获取一次 lock.release() lock.release()
递归锁(RLock)
RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require。直到一个线程所有的acquire都被release,其他的线程才能获得资源。上面的例子如果使
用RLock代替Lock,则不会发生死锁:
import threading if __name__ == '__main__': lock=threading.RLock() lock.acquire() lock.acquire() #可以多次获取,程序顺利执行 lock.release() lock.release()
信号量(Semaphore)
能够并发执行的线程数,超出的线程阻塞,直到有线程运行完成。
Semaphore管理一个内置的计数器,
每当调用acquire()时内置计数器-1;
调用release() 时内置计数器+1;
计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。
import threading,time def sever_help(n): s.acquire() print('%s欢迎用户%d'%(threading.current_thread().getName(),n)) time.sleep(2) s.release() if __name__ == '__main__': s = threading.Semaphore(5) li = [] for i in range(32): t = threading.Thread(target=sever_help,args=(i,)) li.append(t) t.start() for i in li: i.join() print("===结束==")
通过对比互斥锁可以看出,互斥锁就是Semaphore(1)的情况,也完全可以使用后者,但是如果数据必须单独使用,那么用互斥锁效率更高。
事件(Event)
如果某一个线程执行,需要判断另一个线程的状态,就可以使用Event,如:用Event类初始化一个event对象,线程a执行到某一步,设置event.wait(),即线程a阻塞,直到另一个线程设置event.set(),将event
状态设置为True(默认是False)。
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
import threading import time, random def eating(): event.wait() print('去吃饭的路上...') def makeing(): print('做饭中') time.sleep(random.randint(1,2)) print('做好了,快来...') event.set() if __name__ == '__main__': event=threading.Event() t1 = threading.Thread(target=eating) t2 = threading.Thread(target=makeing) t1.start() t2.start() # 做饭中 # 做好了,快来... # 去吃饭的路上...
基本方法:
event.isSet():返回event的状态值; event.wait():如果 event.isSet()==False将阻塞线程; event.set(): 设置event的状态值为True,所有阻塞池的线程激活进入就绪状态, 等待操作系统调度; event.clear():恢复event的状态值为False。
线程队列(queue)
特点:先进先出,
作用:多个线程之间进行通信(作用不大,多进程的队列用处大)
常用方法:
- .get() 获取 无数据时会阻塞
- .set('item') 设置,先设置的数据,先取出
- .empty() 是否为空
基本使用:生成者消费者模型
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
import threading, queue def eating(n): #消费 i = q.get() print('消费者%d吃了第%d份食物' % (n, i)) def making(): #生产 for i in range(1, 11): print('正在制作第%d份食物' % i) time.sleep(1) q.put(i) if __name__ == '__main__': q = queue.Queue() t2 = threading.Thread(target=makeing) t2.start() for i in range(1, 11): t = threading.Thread(target=eating, args=(i,)) t.start()