IPC
进程间是相互独立的,资源无法共享,当一个进程想要把数据给另外一个进程,就需要考虑IPC,可以通过管道(队列)之间通讯
堆栈:先进后出
from multiprocessing import Queue q = Queue(5) # 括号内的参数,表示的是这个队列的储存数 q.put(1) # 添加数据 q.put(2) q.put(3) q.put(4) print(q.full()) # 判断队列是否满了 q.put(5) # q.put(6) # 当队列满了之后,在向里面添加值,不会出现报错,会在原地等待,进入阻塞状态,等待队列中有值被取走 # print(q.__dict__) print(q.get()) # 获取数据 print(q.get()) print(q.get()) print(q.empty()) # 判断队列中的数据取完 print(q.get()) print(q.get()) print(q.get_nowait()) # 取值,没有值不等待直接报错 print(q.get()) # 当队列中的数据被取完之后,再次获取,程序会阻塞,进入阻塞状态,在原地等待,直到有值被添加 """ full empty get_nowait 都不适用于多进程的情况下 刚判断完,就有值被添加或取出,或者还没来得及添加 """
(验证进程间通讯)
子进程放数据 主进程获取数据
两个子进程相互放 取数据
from multiprocessing import Process,Queue def producer(q): q.put('hello world') def consumer(q): print(q.get()) if __name__ == '__main__': q = Queue() p1 = Process(target=producer,args=(q,)) p2 = Process(target=consumer,args=(q,)) p1.start() p2.start()
模型 就是解决某个问题套路
处理数据的一方称之为消费者
例如: 爬虫 生活中到处都是这种模型
饭店 厨师就是生产者 你吃饭的人就是消费者
生产者和消费,处理速度不平衡,一方快一方慢,导致一方需要等待另一方
原本,双方是耦合 在一起,消费必须等待生产者生成完毕在开始处理, 反过来
如果消费消费速度太慢,生产者必须等待其处理完毕才能开始生成下一个数据
将双方分开来.一专门负责生成,一方专门负责处理
这样一来数据就不能直接交互了 双方需要一个共同的容器
生产者完成后放入容器,消费者从容器中取出数据
这样就解决了双发能力不平衡的问题,做的快的一方可以继续做,不需要等待另一方
from multiprocessing import Process,JoinableQueue import random import time def producer(name,food,q): for i in range(5): data = '%s生产了%s%s'%(name,food,i) time.sleep(random.random()) q.put(data) print(data) def consumer(name,q): while True: data = q.get() if data == None: break print('%s吃了%s'%(name,data)) time.sleep(random.random()) q.task_done() # 告诉队列你已经从队列中取出了一个数据 并且处理完毕了 if __name__ == '__main__': q = JoinableQueue() p1 = Process(target=producer,args=('小明','葱油饼',q)) p2 = Process(target=producer,args=('小于','菜包',q)) c1= Process(target=consumer,args=('胖丁',q)) c2 = Process(target=consumer,args=('丫丫',q)) p1.start() p2.start() c1.daemon = True c2.daemon = True c1.start() c2.start() p1.join() p2.join() q.join() # 等到队列中数据全部取出 # q.put(None) # q.put(None)
线程
线程是操作系统最小的运算调度单位,被包含在进程中,一个线程就是一个固定的 执行流程
线程的进程的关系
线程不能单独存在 必须存在于进程中,
进程是一个资源单位,其包含了运行程序所需的所有资源
线程才是真正的执行单位
没有线程,进程中的资源无法被利用起来,所以一个进程至少包含一个线程,称之为主线程
当我们启动一个程序时,操作系统就会自己为这个程序创建一个主线程
线程可以由程序后期开启 ,自己开启线程称之为子线程
进程:资源单位 线程:执行单位 每个进程都自带一个线程,线程才是真正的执行单位,进程只是在线程运行过程中 提供代码运行所需要的资源 为什么要有线程 开进程 1.申请内存空间 耗资源 2."拷贝代码" 耗资源 开线程 一个进程内可以起多个线程,并且线程与线程之间数据是共享的 ps:开启线程的开销要远远小于开启进程的开销
from threading import Thread import time 使用方法一 直接实例化Thread类 def task(name): print('%s is running'%name) time.sleep(1) print('%s is over'%name) # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内 t = Thread(target=task,args=('dada',)) t.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 print('主')
from threading import Thread import time 使用方法二 继承Thread 覆盖run方法 class MyThread(Thread): def __init__(self,name): super().__init__() self.name = name def run(self): print('%s is running'%self.name) time.sleep(3) print('%s is over'%self.name) t = MyThread('cc') t.start() print('主')
其他方法
from threading import Thread,current_thread,active_count import time import os def task(name,i): print('%s is running'%name) # print('子current_thread:',current_thread().name) # print('子',os.getpid()) time.sleep(i) print('%s is over'%name) # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内 t = Thread(target=task,args=('egon',1)) t1 = Thread(target=task,args=('jason',2)) t.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 t1.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 t1.join() # 主线程等待子线程运行完毕 print('当前正在活跃的线程数',active_count()) # 小的代码执行完 线程就已经开启了 print('主') # print('主current_thread:',current_thread().name) # print('主',os.getpid())
守护线程
一个线程可以设置为另一个线程的守护线程
特点: 被守护线程结束后守护线程也随之结束
守护线程会等到所有非守护线程结束后结束 ! 前提是除了主线程之外 还有后别的非守护
当然如果守护线程已经完成任务 立马就结束了
from threading import Thread,current_thread import time def task(i): print(current_thread().name) time.sleep(i) print('GG') # for i in range(3): # t = Thread(target=task,args=(i,)) # t.start() t = Thread(target=task,args=(1,)) t.daemon = True t.start() print('主')
from threading import Thread money = 666 def task(): global money money = 999 t = Thread(target=task) t.start() t.join() print(money)
互斥锁
共享意味着竞争
线程中也存在安全问题,
多线程可以并发执行,一旦并发了并且访问了同一个资源就会有问题
解决方案:还是互斥锁
from threading import Thread,Lock import time n = 100 def task(mutex): global n 上锁 mutex.acquire() tmp = n time.sleep(0.1) n = tmp - 1 释放锁 mutex.release() t_list = [] mutex = Lock() for i in range(100): t = Thread(target=task,args=(mutex,)) t.start() t_list.append(t) for t in t_list: t.join() print(n)
思考
from threading import Thread from multiprocessing import Process import time def foo(): print(123) time.sleep(1) print("end123") def bar(): print(456) time.sleep(3) print("end456") if __name__ == '__main__': t1=Thread(target=foo) t2=Thread(target=bar) t1.daemon=True t1.start() t2.start() print("main-------")