01 进程间通信
""" 队列:先进先出 堆栈:先进后出 """ from multiprocessing import Queue
q = Queue(5) # 括号内可以传参数 表示的是这个队列的最大存储数 # 往队列中添加数据 q.put(1) q.put(2) print(q.full()) # 判断队列是否满了 q.put(3) q.put(4) q.put(5) print(q.full()) # q.put(6) # 当队列满了之后 再放入数据 不会报错 会原地等待 直到队列中有数据被取走(阻塞态)print(q.get()) print(q.get()) print(q.get()) print(q.empty()) # 判断队列中的数据是否取完 print(q.get()) print(q.get()) print(q.empty()) print(q.get_nowait()) # 取值 没有值不等待直接报错 # print(q.get()) # 当队列中的数据被取完之后 再次获取 程序会阻塞 直到有人往队列中放入值 """ full get_nowait empty 都不适用于多进程的情况 """
02 进程间通信IPC机制
from multiprocessing import Queue, Processdef produce(q):q.put('hello,baby~')def consumer(q):print(q.get())if __name__ == '__main__':q = Queue()p = Process(target=produce, args=(q,)) # target(把...作为目标)c = Process(target=consumer, args=(q,))p.start()c.start()
03 生产者消费者模型
""" 生产者:生产/制造数据的 消费者:消费/处理数据的 例子:做包子的,买包子的1.做包子远比买包子的多2.做包子的远比包子的少供需不平衡的问题 """# # multiprocessing:多重处理from multiprocessing import Process, Queue, JoinableQueue import time import randomdef producer(name, food, q):for i in range(1, 10):data = f"{name}做了第{i}份{food}"time.sleep(random.random()) # 随机睡1秒之内 q.put(data)print(data)def consumer(name, q):while True:data = q.get()if data == None: breakprint(f"{name}吃了{data}")time.sleep(random.random())q.task_done()if __name__ == '__main__':q = JoinableQueue()p1 = Process(target=producer, args=('jason', '馒头', q))p2 = Process(target=producer, args=('tank', '生蚝', q))c1 = Process(target=consumer, args=('ryan', q))c2 = Process(target=consumer, args=('ami', q))p1.start()p2.start()c1.daemon = Truec2.daemon = Truec1.start()c2.start()p1.join()p2.join()q.join() # 等队列中数据全部取出
04 线程
什么是线程进程线程其实都是虚拟单位,都是用来帮助我们形象的描述某种事物进程:资源单位线程:执行单位将内存比如成工厂那么进程就相当于是工厂里面的车间而你的线程就相当于是车间里面的流水线ps:每个进程都自带一个线程,线程才是真正的执行单位,进程只是在线程运行过程中提供代码运行所需要的资源为什么要有线程开进程1.申请内存空间 耗资源2."拷贝代码" 耗资源开线程一个进程内可以起多个线程,并且线程与线程之间数据是共享的ps:开启线程的开销要远远小于开启进程的开销如何使用线程
05 创建线程的两种方式
import time from threading import Threaddef run(name):print(f"{name} is running.")time.sleep(2)print(f"{name} is stop.")t = Thread(target=run, args=('ryan',)) # TypeError: run() takes 1 positional argument but 4 were given t.start()
import time from threading import Threadclass MyThread(Thread):def __init__(self, name):super().__init__()self.name = namedef run(self):print(f"{self.name} is running.")time.sleep(2)print(f"{self.name} is over.")t = MyThread('ryan') t.start() print('主')
06 线程对象及其他方法
# 查看线程的活跃数 import os import time from threading import Thread, active_count, current_threaddef task(name, i):print(f"{name} is running")print("主current_thread", current_thread().name)print(os.getpid())time.sleep(i)print(f"{name} is over")t1 = Thread(target=task, args=('ryan', 1)) t2 = Thread(target=task, args=('ryan', 2)) t1.start() t2.start() t2.join() print("当前活跃线程数", active_count()) print('主') print("主current_thread", current_thread().name) print(os.getpid())
07 守护线程
import time from threading import Thread,current_threaddef task(i):print(current_thread().name)time.sleep(i)t = Thread(target=task, args=(1,))t.daemon = True t.start() print('主')
08 线程间通信
from threading import Threadmoney = 666def task():global moneymoney = 999t = Thread(target=task) t.start() t.join() print(money)
09 互斥锁
from threading import Thread, Lock import timen = 100def task(mutex):global nmutex.acquire() # 获得tmp = ntime.sleep(0.1)n = tmp - 1mutex.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)
10 小例子
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=Truet1.start()t2.start()print("main-------")
