python之并发编程

python之并发编程

  • 线程的创建方式
    • 线程的创建方式(方法包装)
    • 线程的创建方式(类包装)
    • join()【让主线程等待子线程结束】
    • 守护线程【主线程结束,子线程就结束】
    • 多线程操作同一个对象(未使用线程同步)
    • 多线程操作同一个对象(增加互斥锁,使用线程同步)
    • 死锁案例
    • 信号量使用
    • 事件(Event)
  • 生产者消费者模型
  • 进程
    • 方法模式创建进程
    • 类模式创建进程
    • Queue实现进程间通信
    • Pipe实现进程间通信
    • Manager管理器
  • 进程池(Pool)
    • 进程池使用案例
    • 使用with管理进程池
  • 协程是什么
    • 不使用协程执行多个任务
    • 使用yield协程,实现任务切换
    • asyncio实现协程(重点)

线程的创建方式

Python的标准库提供了两个模块: _thread 和 threading , _thread 是低级模块, threading 是高级模块,对 _thread 进行了封装。绝大多数情况下,我们只需要使用 threading 这个高级模块。
线程的创建可以通过分为两种方式:

  1. 方法包装
  2. 类包装
    线程的执行统一通过 start() 方法

线程的创建方式(方法包装)

def func1(name):print(f"线程{name},start")  #formatfor i in range(3):print(f"线程{name},{i}")sleep(3)print(f"线程{name},end")if __name__ == '__main__':print("主线程,start")#创建线程t1 = Thread(target=func1,args=("t1",))t2 = Thread(target=func1,args=("t2",))#启动线程t1.start()t2.start()print("主线程,end")
运行结果可能会出现换行问题,是因为多个线程抢夺控制台
输出的IO流。

线程的创建方式(类包装)

class MyThread(Thread):def __init__(self,name):Thread.__init__(self)self.name = namedef run(self):print(f"线程{self.name},start")  # formatfor i in range(3):print(f"线程{self.name},{i}")sleep(3)print(f"线程{self.name},end")if __name__ == '__main__':print("主线程,start")#创建线程t1 = MyThread("t1")t2 = MyThread("t2")#启动线程t1.start()t2.start()print("主线程,end")

join()【让主线程等待子线程结束】

之前的代码,主线程不会等待子线程结束。
如果需要等待子线程结束后,再结束主线程,可使用join()方法。

def func1(name):for i in range(3):print(f"thread:{name} :{i}")sleep(1)if __name__ == '__main__':print("主线程,start")#创建线程t1 = Thread(target=func1,args=("t1",))t2 = Thread(target=func1,args=("t2",))#启动线程t1.start()t2.start()#主线程会等待t1,t2结束后,再往下执行t1.join()t2.join()print("主线程,end")

守护线程【主线程结束,子线程就结束】

class MyThread(Thread):def __init__(self,name):Thread.__init__(self)self.name =namedef run(self):for i in range(3):print(f"thread:{self.name} :{i}")sleep(1)if __name__ == '__main__':print("主线程,start")#创建线程(类的方式)t1 = MyThread('t1')#t1设置为守护线程t1.daemon = True# t1.setDaemon(True)#启动线程t1.start()print("主线程,end")

多线程操作同一个对象(未使用线程同步)

class Account:def __init__(self,money,name):self.money = moneyself.name = name#模拟提款的操作
class Drawing(Thread):def __init__(self,drawingNum,account):Thread.__init__(self)self.drawingNum = drawingNumself.account = accountself.expenseTotal = 0def run(self):if self.account.money<self.drawingNum:returnsleep(1) #判断完可以取钱,则阻塞。就是为了测试发生冲突问题self.account.money -=self.drawingNumself.expenseTotal += self.drawingNumprint(f"账户:{self.account.name},余额是:{self.account.money}")print(f"账户:{self.account.name},总共取了:{self.expenseTotal}")if __name__ == '__main__':a1 = Account(100,"gaoqi")draw1 = Drawing(80,a1)  #定义一个取钱的线程draw2 = Drawing(80,a1)  #定义一个取钱的线程draw1.start()draw2.start()

多线程操作同一个对象(增加互斥锁,使用线程同步)

class Account:def __init__(self,money,name):self.money = moneyself.name = name#模拟提款的操作
class Drawing(Thread):def __init__(self,drawingNum,account):Thread.__init__(self)self.drawingNum = drawingNumself.account = accountself.expenseTotal = 0def run(self):lock1.acquire()if self.account.money<self.drawingNum:print("账户余额不足!")returnsleep(1) #判断完可以取钱,则阻塞。就是为了测试发生冲突问题self.account.money -=self.drawingNumself.expenseTotal += self.drawingNumlock1.release()print(f"账户:{self.account.name},余额是:{self.account.money}")print(f"账户:{self.account.name},总共取了:{self.expenseTotal}")if __name__ == '__main__':a1 = Account(1000,"gaoqi")lock1 = Lock()draw1 = Drawing(80,a1)  #定义一个取钱的线程draw2 = Drawing(80,a1)  #定义一个取钱的线程draw1.start()draw2.start()

死锁案例

def fun1():lock1.acquire()print('fun1拿到菜刀')sleep(2)lock2.acquire()print('fun1拿到锅')lock2.release()print('fun1释放锅')lock1.release()print('fun1释放菜刀')def fun2():lock2.acquire()print('fun2拿到锅')lock1.acquire()print('fun2拿到菜刀')lock1.release()print('fun2释放菜刀')lock2.release()print('fun2释放锅')if __name__ == '__main__':lock1 = Lock()lock2 = Lock()t1 = Thread(target=fun1)t2 = Thread(target=fun2)t1.start()t2.start()

信号量使用

def home(name,se):se.acquire()print(f"{name}进入房间")sleep(3)print(f"****{name}走出房间")se.release()if __name__ == '__main__':se = Semaphore(5)   #信号量对象for i in range(7):t = Thread(target=home,args=(f"tom{i}",se))t.start()

事件(Event)

事件Event主要用于唤醒正在阻塞等待状态的线程;

Event 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在初始情况下,event 对象中的信号标志被设置假。如果有线程等待一个 event 对象,而这个 event 对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个 event 对象的信号标志设置为真,它将唤醒所有等待个 event 对象的线程。如果一个线程等待一个已经被设置为真的 event 对象,那么它将忽略这个事件,继续执行

在这里插入图片描述

def chihuoguo(name):#等待事件,进入等待阻塞状态print(f'{name}已经启动')print(f'小伙伴{name}已经进入就餐状态!')time.sleep(1)event.wait()# 收到事件后进入运行状态print(f'{name}收到通知了.' )print(f'小伙伴{name}开始吃咯!')if __name__ == '__main__':event = threading.Event()# 创建新线程thread1 = threading.Thread(target=chihuoguo, args=("tom", ))thread2 = threading.Thread(target=chihuoguo, args=("cherry", ))# 开启线程thread1.start()thread2.start()time.sleep(10)# 发送事件通知print('---->>>主线程通知小伙伴开吃咯!')event.set()

生产者消费者模型

从一个线程向另一个线程发送数据最安全的方式可能就是使用queue 库中的队列了。创建一个被多个线程共享的 Queue 对象,这些线程通过使用 put() 和 get() 操作来向队列中添加或者删除元素。Queue 对象已经包含了必要的锁,所以你可以通过它在多个线程间多安全地共享数据。

def producer():num = 1while True:if queue.qsize()<5:print(f"生产{num}号,大馒头")queue.put(f"大馒头:{num}号")num +=1else:print("馒头框满了,等待来人消费啊!")sleep(1)def consumer():while True:print(f"获取馒头:{queue.get()}")sleep(1)if __name__ == '__main__':queue = Queue()t1 = Thread(target=producer)t2 = Thread(target=consumer)t1.start()t2.start()

进程

方法模式创建进程

def fun1(name):print(f"当前进程ID:{os.getpid()}")print(f"父进程ID:{os.getppid()}")print(f"Process:{name},start")sleep(3)print(f"Process:{name},end")#windows上多进程实现的bug。如果不加main的限制,就会无限制的创建子进程,从而报错。
if __name__ == '__main__':print("当前进程ID:",os.getpid())#创建进程p1 = Process(target=fun1,args=("p1",))p2 = Process(target=fun1, args=("p2",))#启动进程p1.start()p2.start()

类模式创建进程

class MyProcess(Process):def __init__(self,name):Process.__init__(self)self.name = namedef run(self):print(f"Process:{self.name},start")sleep(3)print(f"Process:{self.name},end")if __name__ == '__main__':#创建进程p1 = MyProcess("p1")p2 = MyProcess("p2")p1.start()p2.start()

Queue实现进程间通信

前面讲解了使用 Queue 模块中的 Queue 类实现线程间通信,但要实现进程间通信,需要使用 multiprocessing 模块中的 Queue 类。
简单的理解 Queue 实现进程间通信的方式,就是使用了操作系统给开辟的一个队列空间,各个进程可以把数据放到该队列中,当然也可以从队列中把自己需要的信息取走。

#coding=utf-8
from multiprocessing import Process, Queue
from time import sleepclass MyProcess(Process):def __init__(self,name,mq):Process.__init__(self)self.name = nameself.mq = mqdef run(self):print(f"Process:{self.name},start")print(f"get Data:{mq.get()}")sleep(2)self.mq.put(f"new_data:{self.name}")print(f"Process:{self.name},end")if __name__ == '__main__':mq = Queue()mq.put("1")mq.put("2")mq.put("3")#进程列表p_list = []for i in range(3):p = MyProcess(f"p{i}",mq)p_list.append(p)for p in p_list:p.start()for p in p_list:p.join()print(mq.get())print(mq.get())print(mq.get())

Pipe实现进程间通信

Pipe 直译过来的意思是“管”或“管道”,和实际生活中的管(管道)是非常类似的。
Pipe方法返回(conn1, conn2)代表一个管道的两个端。Pipe方法有duplex参数,如果duplex参数为True(默认值),
那么这个参数是全双工模式,也就是说conn1和conn2均可收发。若duplex为False,conn1只负责接收消息,conn2只负责
发送消息。send和recv方法分别是发送和接受消息的方法。例如,在全双工模式下,可以调用conn1.send发送消息,
conn1.recv接收消息。如果没有消息可接收,recv方法会一直阻塞。如果管道已经被关闭,那么recv方法会抛出EOFError。

#coding=utf-8
import multiprocessing
from time import sleepdef func1(conn1):sub_info = "Hello!"print(f"进程1--{multiprocessing.current_process().pid}发送数据:{sub_info}")sleep(1)conn1.send(sub_info)print(f"来自进程2:{conn1.recv()}")sleep(1)
def func2(conn2):sub_info = "你好!"print(f"进程2--{multiprocessing.current_process().pid}发送数据:{sub_info}")sleep(1)conn2.send(sub_info)print(f"来自进程1:{conn2.recv()}")sleep(1)if __name__ == '__main__':#创建管道conn1,conn2 = multiprocessing.Pipe()# 创建子进程process1 = multiprocessing.Process(target=func1,args=(conn1,))process2 = multiprocessing.Process(target=func2,args=(conn2,))# 启动子进程process1.start()process2.start()

Manager管理器

def func(name,m_list,m_dict):m_dict['name'] = '123'm_list.append('你好')if __name__ == "__main__":with Manager() as mgr:m_list = mgr.list()m_dict = mgr.dict()m_list.append('Hello!!')#两个进程不能直接互相使用对象,需要互相传递p1 = Process(target=func,args=('p1',m_list,m_dict))p1.start()p1.join()   #等p1进程结束,主进程继续执行print(f"主进程:{m_list}")print(f"主进程:{m_dict}")

进程池(Pool)

Python提供了更好的管理多个进程的方式,就是使用进程池
进程池可以提供指定数量的进程给用户使用,即当有新的请求提交到进程池中时,如果池未满,则会创建一个新的进程用来执行该请求;反之,如果池中的进程数已经达到规定最大值,那么该请求就会等待,只要池中有进程空闲下来,该请求就能得到执行。
在这里插入图片描述

进程池使用案例

def func1(name):print(f"当前进程的ID:{os.getpid()},{name}")sleep(2)return namedef func2(args):print(args)if __name__ == "__main__":pool = Pool(5)pool.apply_async(func = func1,args=('sxt1',),callback=func2)pool.apply_async(func = func1,args=('sxt2',),callback=func2)pool.apply_async(func = func1,args=('sxt3',),callback=func2)pool.apply_async(func = func1,args=('sxt4',))pool.apply_async(func = func1,args=('sxt5',))pool.apply_async(func = func1,args=('sxt6',))pool.apply_async(func = func1,args=('sxt7',))pool.apply_async(func = func1,args=('sxt8',))pool.close()pool.join()

使用with管理进程池

def func1(name):print(f"当前进程的ID:{os.getpid()},{name}")sleep(2)return nameif __name__ == "__main__":with Pool(5) as pool:args = pool.map(func1,('sxt1,','sxt2,','sxt3,','sxt4,','sxt5,','sxt6,','sxt7,','sxt8,'))for a in args:print(a)

协程是什么

协程,Coroutines,也叫作纤程(Fiber)
协程,全称是“协同程序”,用来实现任务协作。是一种在线程中,比线程更加轻量级的存在,由程序员自己写程序来管理。
当出现IO阻塞时,CPU一直等待IO返回,处于空转状态。这时候用协程,可以执行其他任务。当IO返回结果后,再回来处理数据。充分利用了IO等待的时间,提高了效率。一个故事说明进程、线程、协程的关系

乔布斯想开工厂生产手机,费劲力气,制作一条生产线,这个生产线上有很多的器件以及材料。一条生产线就是一个进程。
只有生产线是不够的,所以找五个工人来进行生产,这个工人能够利用这些材料最终一步步的将手机做出来,这五个工人就
是五个线程。
为了提高生产率,想到3种办法:
1 一条生产线上多招些工人,一起来做手机,这样效率是成倍増长,即单进程多线程方式
2 多条生产线,每个生产线上多个工人,即多进程多线程
乔布斯深入一线发现工人不是那么忙,有很多等待时间。于是规定:如果某个员工在等待生
产线某个零件生产时 ,不要闲着,干点其他工作。也就是说:如果一个线程等待某些条件,
可以充分利用这个时间去做其它事情,这就是:协程方式。

不使用协程执行多个任务

def func1():for i in range(3):print(f'北京:第{i}次打印啦')time.sleep(1)return "func1执行完毕"
def func2():for k in range(3):print(f'上海:第{k}次打印了' )time.sleep(1)return "func2执行完毕"def main():func1()func2()
if __name__ == '__main__':start_time = time.time()main()end_time = time.time()print(f"耗时{end_time-start_time}")   #不使用协程,耗时6秒

使用yield协程,实现任务切换

def func1():for i in range(3):print(f'北京:第{i}次打印啦')yield  # 只要方法包含了yield,就变成一个生成器time.sleep(1)
def func2():g = func1()    #func1是一个生成器,func1()就不会直接调用,需要通过next()print(type(g))for k in range(3):print(f'上海:第{k}次打印了' )next(g)   #继续执行func1的代码time.sleep(1)if __name__ == '__main__':#有了yield,我们实现了两个任务的切换+保存状态start_time = time.time()func2()end_time = time.time()print(f"耗时{end_time-start_time}")   #耗时5.0秒,效率差别不大

asyncio实现协程(重点)

  1. 常的函数执行时是不会中断的,所以你要写一个能够中断的函数,就需要加 async.async 用来声明一个函数为异步函数,异步函数的特点是能在函数执行过程中挂起,去执行其他异步函数,等到挂起条件(假设挂起条件是 sleep(5) )消失后,也就是5秒到了再回来执行
  2. await 用来用来声明程序挂起,比如异步程序执行到某一步时需要等待的时间很长,就将此挂起,去执行其他的异步程序。
  3. asyncio 是python3.5之后的协程模块,是python实现并发重要的包,这个包使用事件循环驱动实现并发。
async def func1():     #async表示方法是异步的for i in range(3):print(f'北京:第{i}次打印啦')await asyncio.sleep(1)return "func1执行完毕"
async def func2():for k in range(3):print(f'上海:第{k}次打印了' )await asyncio.sleep(1)return "func2执行完毕"
async def main():res = await asyncio.gather(func1(), func2())#await异步执行func1方法#返回值为函数的返回值列表print(res)if __name__ == '__main__':start_time = time.time()asyncio.run(main())end_time = time.time()print(f"耗时{end_time-start_time}")   #耗时3秒,效率极大提高

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/687662.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

SSIM(Structural Similarity),结构相似性及MATLAB实现

参考文献 Wang, Zhou; Bovik, A.C.; Sheikh, H.R.; Simoncelli, E.P. (2004-04-01). “Image quality assessment: from error visibility to structural similarity”. IEEE Transactions on Image Processing. 13 (4): 600–612. Bibcode:2004ITIP…13…600W. CiteSeerX 10.…

数字型隔离器ISO121x的用法

目录 概述 1 认识ISO121x 1.1 简介 1.2 特性 1.3 应用领域 2 ISO121x芯片结构 2.1 ISO1211引脚介绍 2.2 ISO1211的通用应用电路 2.3 Layout Example 3 应用范例 3.1 TI提供的评估板 3.2 评估板的原理图电路 概述 本文主要介绍ISO121x的相关特性&#xff0c;以及其…

9.spring-图书管理系统

文章目录 1.开发项目流程1.1开发开发1.2数据库的设计 2.MySQL数据库相关代码3.构造图书结构3.1用户登录3.2图书列表3.3图书添加3.4图书删除3.4.1批量删除 3.5图书查询(翻页) 4.页面展示4.1登录页面4.2列表页面4.3增加图书页面4.4修改图书信息页面 5.功能展示5.1增加图书信息5.2…

MySQL·表的内外连接

目录 表的内连和外连 内连接 案例1&#xff1a;显示SMITH的名字和部门名 外连接 左外连接 案例2&#xff1a; 查询所有学生的成绩&#xff0c;如果这个学生没有成绩&#xff0c;也要将学生的个人信息显示出来 右外连接 案例3&#xff1a;对stu表和exam表联合查询&#…

【Linux】什么是进程?

一个正在执行的程序&#xff0c;我们称之为进程。 然后我们来顺着一条线来思考。 操作系统底层是用C语言编写的&#xff0c;而我们的进程&#xff0c;它会有各种属性&#xff0c;那么各种属性就可以用一个结构体来对进程的各个属性进行描述&#xff0c;然后这个结构体里面&…

linux安装clamav病毒扫描与删除

ClamAV介绍 ClamAV是Linux操作系统一款免费的杀毒工具&#xff0c;通过命令执行病毒库升级、查找病毒和删除病毒。 安装ClamAV 方法一&#x1f4a1; Tips&#xff1a;在CentOS操作系统上安装ClamAV&#xff0c;请分别执行以下命令 yum install epel-release -y yum install cla…

Python 运筹优化11 BernoulliBandit 解读

说明 以广告点击的案例继续MultiArmed Bandit的学习。 内容 1 概要 样例假设存在5个广告&#xff0c;通过伯努利分布来模拟广告的点击可能。 adA BernoulliBandit(0.004) adB BernoulliBandit(0.016) adC BernoulliBandit(0.02) adD BernoulliBandit(0.028) adE Bern…

sipeed 的 MaixCam UART操作

发现问题 根据sipeed MaixCam官方文档 使用MaixVision会报错。 正确的接线 1&#xff0c;usb转ttl的RX和TX与sipeed MaixCam官方赠送的usb转接头反向连接&#xff0c;GND互相连接。 2&#xff0c;再用一根tpyc-c为其供电。 连接WiFi路由器 MaixCam液晶屏输入WiFi名称和密…

requestAnimationFrame请求动画帧

一、前言 在Web应用中&#xff0c;实现动画效果的方法比较多&#xff1a; CSS3&#xff1a;Transition&#xff08;过度&#xff09; / Animation&#xff08;动画&#xff09; HTML5&#xff1a;Canvas JavaScript&#xff1a;setInterval&#xff08;定时器&#xff09; /…

腾讯云服务器部署前后端服务

服务器&#xff1a;OpenCloudOS &#xff08;兼容centos8&#xff09; 后端&#xff1a;javaSpringboot 前端&#xff1a;Vue 下载jdk 1&#xff09;下载jdk11 wget https://download.java.net/openjdk/jdk11/ri/openjdk-1128_linux-x64_bin.tar.gz 2&#xff09;解压jdk …

印象笔记大师:揭秘高效管理的终极技巧,让你的信息整理事半功倍!

印象笔记&#xff08;Evernote&#xff09;是一款功能强大的笔记应用&#xff0c;它可以帮助用户记录、整理和管理各种信息。无论是工作、学习还是日常生活&#xff0c;印象笔记都能为我们提供便捷的服务。本文将全面详细地介绍印象笔记的使用技巧&#xff0c;包括基本介绍、主…

Gartner发布降低企业软件供应链安全风险指南:全球软件供应链相关法规、指南以及企业需要开展的三个方面工作

软件供应链攻击呈三位数增长&#xff0c;但很少有企业机构采取措施对这些复杂攻击的风险进行评估。安全和风险管理领导者可参考本文&#xff0c;采用三种实践来检测和预防攻击&#xff0c;保护企业机构的安全。 主要发现 虽然软件供应链攻击频繁发生&#xff0c;但其安全评估尚…