Day 33 并发编程3

生产者消费者模型

模型就是解决某个问题的固定方法或者套路

要解决什么问题

• 生产者:泛指生产数据的一方
• 消费者:负责把处理数据的一方

案例:

​ 食堂饭店是生产者

​ 吃饭的人是消费者

他们之间的问题:

​ 效率低,因为双方的处理速度不一致,双方需要相互等待

具体的解决方案

1. 先将双方解开耦合,让不同的进程负责不同的任务
2. 提供一个共享容器,来平衡双方的能力,之所以用进程队列是因为队列可以在进程间共享

案例

from multiprocessing import Process,Queue
import requests,time,random,re,osdef product(urls,q):i=1for url in urls:respone=requests.get(url)text=respone.text# time.sleep(random.random())q.put(text)print(os.getpid(),f'got NO{i} info')i+=1def customer(q):i=1while True:text=q.get()time.sleep(random.random())res=re.findall('src=//(.*?) width', text)print(f'got NO{i} img')i++1if __name__ == '__main__':urls = ["http://www.baidu.com","http://www.baidu.com","http://www.baidu.com","http://www.baidu.com",]q=Queue()p=Process(target=product,args=(urls,q))p.start()c=Process(target=customer,args=(q,))c.start()

问题:customer进程不知道什么时候结束

joinableQueue继承自Queue,用法一致

增加了join和taskDone

join是一个阻塞函数,会阻塞直到taskdone的调用次数等于存入的元素个数,可以用于表示队列任务处理完成

from multiprocessing import Process,JoinableQueue
import requests,time,random,re,osdef product(q,name):for i in range(5):hot_dog=f'hot_dog{i+1} from {name}'time.sleep(random.random())q.put(hot_dog)print(f'{name} make hot_dog{i+1}')def customer(q):while True:hot_dog=q.get()time.sleep(random.random())print(f'eat {hot_dog}')q.task_done()if __name__ == '__main__':q=JoinableQueue()p1=Process(target=product,args=(q,'shanghai'))p1.start()p2=Process(target=product,args=(q,'beijing'))p2.start()c=Process(target=customer,args=(q,))c.start()p1.join()p2.join()q.join()c.terminate()

用途

常用来做流量削峰,保证服务不会因为高并发而崩溃

多线程

什么是线程

进程是操作系统可以调度和资源分配的基本单位,是一个资源单位,其中包含了运行这个程序所需的资源

线程是操作系统可以运算调度的最小单位,是真正的执行单位,一个线程就是一条固定的控制流程

一个进程可以包含多个线程,统一进程中的线程共享进程内的资源

系统会为每一个进程自动创建一条线程,称之为主线程,后续通过代码开启的线程称之为子线程

进程对比线程

进程是一个资源单位 而线程是执行单位

创建进程的开销大于线程

多个进程之间的内存是相互隔离的,而线程是共享进程内的所有资源

进程间是竞争关系,而线程间是合作关系

开启线程也需要消耗资源

进程之间有子父级关系

为什么用线程

1. 有多个任务要并发处理
2. 当要并发处理的任务有很多的时候,不能使用进程,进程资源开销大

使用线程

方式一:直接实例化thread类

方式二:继承thread类

from threading import Threaddef task():print('son run')t=Thread(target=task)   #子线程不需要重新读取主线程的代码 所以开启线程的代码可以随便放在需要的位置
t.start()
print('over')
# 线程开启速度快 所以先打印son run 后打印overclass MyThread(Thread):def run(self) -> None:print(' son run')t=MyThread()
t.start()
print('over')

线程安全问题

线程并发访问了同一资源一定会产生安全问题,解决方案和进程一样,就是给公共资源加锁

from threading import Thread,Lock
import timel=Lock()
a=10def task():global al.acquire()time.sleep(1)a-=1print(a)l.release()for i in range(10):t=Thread(target=task)t.start()print(f'{i} work')
# 瞬间开启10个线程,但是同一时间只有1个线程可以执行函数代码

守护线程

一个线程a设置为b的守护线程,a会随着b的结束而结束

默认情况下主线程代码执行完毕,也会等待所有子线程执行完毕后才能结束,因为多个线程之间是协作关系

from threading import Thread,Lock
import timedef task1():print('this is taks1')time.sleep(5)print('task1 going to die')def task2():print('this is task2')time.sleep(4)print('task2 not dead')print('main will be dead in 3 seconds')
t1=Thread(target=task1)
t1.daemon=True
t1.start()
t2=Thread(target=task2)
t2.daemon=False
t2.start()time.sleep(1)
print('main dead')

线程中的常用方法

from threading import  Thread,currentThread,enumerate,activeCount
import time# t = Thread()
# t.start()
# t.join()
# t.is_alive()
# t.isAlive()
# t.ident  # 线程标识符   id
# t.daemon# 获取当前线程对象
# print(currentThread())
# t = Thread(target=lambda :print(currentThread()))
# t.start()t = Thread(target=lambda :time.sleep(1))
t.start()t = Thread(target=lambda :time.sleep(1))
t.start()
t.join()
# 获取正在运行的所有线程对象  是一个列表
print(enumerate())# 存活的线程数量
print(activeCount())