Python中的并发编程

专治羊毛辣子

我们常听到电脑是单核，多核的，那么这是什么意思呢？单核cpu指的是一次只能运行一个任务。这个任务指的就是进程，进程就好比工厂中的车间。一个车间可以有很多工人，又要引入一个新的概念，线程，线程好比一个个工人，他们共同完成一个任务。这就意味着一个进程可以包括很多线程。进程空间是被线程共享的。这时候要引入一个新的概念互斥锁，好比工厂中一些特殊的房间，当在被使用时就要上锁以防止其他人的访问，当不用时又要解锁，线程也是同理。进程的狭义定义：一个正在运行的程序在操作系统中被视为一个进程。到现在可能还是有一些疑问，当我们使用单核电脑，理论上只能运行一个进程，但是同时运行多个应用时，似乎感觉都在同时运行，这是为什么呢？因为存在“时间片轮转法”的调度算法，简单来说，就是进程间交替的频率很高，看起来就像同时进行。并发与并行并行：同时进行，只有具备多个cpu才能实现并行并发：是伪并行，看起来是同时运行并行与并发都表示cpu有同时处理多任务的能力。 到底是多任务并行还是并发是我们用Python代码把握不了的阻塞状态：正在执行的进程由于某些事件而暂时无法执行，进程受到阻塞，进入就绪状态等待系统调用对于阻塞有不同的处理方式：同步和异步同步：遇到阻塞状态就等着，必须等到方法调用返回后，才能继续后续方法的执行异步：遇到阻塞状态就去处理后续其它操作(既可以基于并行也可以基于并发，通常是基于并发)进程通常分为以下几个状态就绪状态    进程已经准备好，已经分配到资源。执行/运行状态    进程处于就绪状态被调度后，进程进入执行状态阻塞(Blocked)状态(耗时操作)         正在执行的进程由于某些事件而暂时无法执行，进程受到阻塞，进入就绪状态等待系统调用。终止状态     进程结束，或出现错误，或被系统终止，进入终止状态。无法再执行总结并行，并发，同步，异步关系：并行并发是指操作系统处理多个任务的能力。同步异步是指处理某一个任务的过程中，遇到阻塞后不同的处理行为。因为异步处理更加高效，所以接下来是异步的具体操作：进程，线程，协程。Python中进程的实现multiprocess包：是管理进程的包，几乎包含进程有关的所有子模块。Process模块：创建进程的模块，可以在主进程中创建子进程创建进程frommultiprocessingimportProcessdeff():print("我是子进程的任务")if__name__=='__main__':#程序开始入口print('主进程开始')#创建一个进程p，给进程绑定任务p=Process(target=f)#target是固有参数#启动创建好的进程p.start()print('主进程结束')接下来观察主进程与子进程的关系 os.getpod()获取自己的进程ID号os.getppid()获取父进程的ID号 importosfrommultiprocessingimportProcessfromtimeimportsleepdeff():print('我是子进程')sleep(1)print(f'子进程ID号：{os.getpid()}')print(f'该子进程的父进程ID号：{os.getppid()}')if__name__=='__main__':print(f'主进程开始执行！主进程的ID号：{os.getpid()}')#创建一个进程p，绑定一个任务p=Process(target=f)#启动创建好的进程p.start()print('主进程结束') 可以借此观察主进程和子进程关系如何给子进程传参frommultiprocessingimportProcessdeff(n1,n2):print(f"我是子进程的任务{n1,n2}")if__name__=='__main__':#程序开始入口print('主进程开始')#创建一个进程p，给进程绑定任务p=Process(target=f,args=(1,2))#target，args是固有参数#启动创建好的进程p.start()print('主进程结束')下面用同步，异步代码分别申请三个网址的时间，来分析相关性能模拟实现同步：importtimedefget_request(url):print(f'正在请求网址数据；{url}')time.sleep(2)print(f'请求结束f{url}')if__name__=='__main__':#程序开始入口start=time.time()urls=['www.1.com','www,2,com','www.3.com']forurlinurls:get_request(url)print(f'总耗时{time.time()-start}') 因为每申请一个网址就sleep两秒，申请了三个网址。所以花了六秒 那么异步操作呢：importtimefrommultiprocessingimportProcessdefget_request(url):print(f'正在请求网址数据；{url}')time.sleep(2)print(f'请求结束f{url}')if__name__=='__main__':#程序开始入口start=time.time()urls=['www.1.com','www,2,com','www.3.com']ps=[]#储存创建好的子进程forurlinurls:#创建子进程，交替执行p=Process(target=get_request,args=(url,))#args=(url,)必须要加,因为是元组ps.append(p)#启动子进程p.start()forpinps:#使的每一个子进程都执行了join操作，意味着：主进程需要等待所有执行了join操作的子进程结束后再结束p.join()print(f'总耗时:{time.time()-start}')在循环里面创建进程，再在另一个循环中进行join操作，不能把join操作与创建进程放在一个循环里，不然会失去异步操作。注意进程间的数据是隔离的p.join可以使主进程在子进程结束后结束可以看出来异步操作明显更加高效 只需了解：也可以用继承面向对象的方式创建进程 守护进程(了解)主进程结束后子进程必须强制结束语法： p.daemon=TrueimporttimefrommultiprocessingimportProcessdefget_request(url):print(f'正在请求网址数据；{url}')time.sleep(2)print(f'请求结束f{url}')if__name__=='__main__':#程序开始入口start=time.time()p=Process(target=get_request,args=('www.1.com'))#将当前p子进程设置为守护进程p.daemon=True#该操作必须放在子进程启动操作之前p.start()print('主进程结束')可以看出来主进程结束后子进程直接强制结束了进程同步锁当我们让几个进程中并发处理，但是它们之间没有顺序，一旦开始也不受我们控制importosimporttimeimportrandomfrommultiprocessingimportProcessdefwork(n):print(f'{n,os.getpid()}isrunning')time.sleep(random.random())print(f'{n,os.getpid()}isdone')if__name__=='__main__':foriinrange(5):p=Process(target=work,args=(i,))p.start() 可以看出来进程进行的顺序并不是按创建顺序进行的 这样不用锁会在后续实现模拟抢票功能中出现错误，当只剩一张票。A用户(进程)执行到一部分可能因为阻塞原因就突然去执行另一个进程(其他用户)，会导致抢票出现混乱。所以应该给一个进程上锁，当一个进程执行完之后再去执行另一个进程。下面将简单模拟抢票机制frommultiprocessingimportLockfrommultiprocessingimportProcessimporttime,json,randomdefsearch():#查询文件中的剩余票数fp=open('./db.txt','r')dic=json.load(fp)#反序列化，将文件中json数据转换为字典print(f'剩余车票数量为：{dic["count"]}')defget():fp=open('./db.txt','r')dic=json.load(fp)time.sleep(0.1)ifdic['count']>0:time.sleep(0.2)dic['count']-=1#一次只能购买一次票time.sleep(0.1)#购买车票后，余票数量变化，重新写回文件中json.dump(dic,open('./db.txt','w'))print('购票成功')deftask(lock):lock.acquire()#上锁search()#先查询get()#后购买lock.release()#解锁if__name__=='__main__':lock=Lock()foriinrange(3):#创建三个子进程p=Process(target=task,args=(lock,))p.start()这里用苹果系统可能执行不成功，因为其与Windows底层实现锁机制有区别 线程线程是系统调度资源的最小单位，同一个进程中的多个线程是共享进程中的资源的。创建线程的成本比创建进程消耗更低，所以在异步操作中操作线程比操作进程更多一些。多线程既可以实现并行，也可以实现并发，但大多数情况下实现的是并行。但在Python中只能实现并发，这是历史遗留问题(GLL锁)。 创建线程除了调用的模块不一样，基本一样fromthreadingimportThreaddeff(num):print(f'num的值是：{num}')if__name__=='__main__':#创建好一个子进程(在主进程中创建)t=Thread(target=f,args=(1,))t.start()也可以面向对象的方式创建线程线程中调用join方法跟进程一样 守护线程fromthreadingimportThreadimporttimedefwork():time.sleep(1)print('子进程正在执行')if__name__=='__main__':t=Thread(target=work)t.daemon=Truet.start()print('主进程结束')线程共享同一进程资源fromthreadingimportThreadimporttimedefwork():globalnn=0if__name__=='__main__':n=1#全局变量t=Thread(target=work)t.start()print(n) 可以看出来子线程共享主进程的变量n，子线程修改n，主进程也跟着修改了。多线程使用运用的是面向对象的方式创建的线程fromthreadingimportThreadimporttimeclassMyThread(Thread):def__init__(self):super().__init__()defrun(self):print('当前子线程正在执行')time.sleep(2)print(('当前子线程执行结束'))if__name__=='__main__':start=time.time()ts=[]foriinrange(3):t=MyThread()t.start()ts.append(t)fortints:t.join()print(f'总耗时：{time.time()-start}')计算密集型任务：任务中大多数是进行计算的经过验证在运行计算密集型任务时，多进程更加高效，因为可以享用多核的优势IO密集型任务：任务中大多数是阻IOIO密集型任务最好使用多线程，因为创建线程的代价更小一些在爬虫中更多使用多线程，因为在爬虫中会出现反复发请求，返回数据的情况多线程上锁fromthreadingimportThread,Lockimporttime,randomdefwork():globalnlock.acquire()temp=ntime.sleep(random.random())n=temp-1lock.release()if__name__=='__main__':n=10ts=[]lock=Lock()foriinrange(10):t=Thread(target=work)t.start()ts.append(t)fortints:t.join()print('全局变量n的值为',n)线程池在内存池中的池化技术是一种内存管理技术，其向操作系统申请一块较大的内存，当程序需要内存时直接向内存池中申请，而不是向操作系统中申请，当程序释放内存时，并不是直接返回给操作系统，而是返还给内存池。大大提高任务效率线程池在爬虫中也十分重要importtimefrommultiprocessing.dummyimportPoolstart=time.time()urls=['www.1.com','www.2.com','www.3.com','www.4.com','www.5.com']defget_request(url):print(f'正在请求数据：{url}')time.sleep(2)print(f'请求结束{url}')pool=Pool(5)#创建线程池对象，池中有五个线程对象,根据需求创建多少个线程对象pool.map(get_request,urls)#表示urls中每个参数依次传给get_request函数print(f'总耗时：{time.time()-start}')pool.close()map中已经实现了join  可以看出来执行五次依旧大约是两秒