1. course
1.进程创建的两种方式
开启进程的第一种方式:
from multiprocessing import Process import random import time def task(name): print(f'{name} is running') time.sleep(random.randint(1, 3)) print(f'{name} is gone') if __name__ == '__main__': # 在windows环境下, 开启进程必须在 __name__ == '__main__' 下面 p = Process(target=task, args=('常鑫')) # 创建一个进程对象 p.start() ''' 只是想操作系统发出一个开辟子进程的信号,然后就执行下一行 这个信号操作系统收到后,会从内存中开辟一个子进程空间,然后将主进程所有数据copy加载到子进程,然后再调用cpu去执行开辟子进程的开销很大 ''' print('开始') time.sleep(2) # 所以永远先执行主进程的代码开启进程的第二种方式:
from multiprocessing import Process import random import time class MyProcess(Process): def __init__(self, name): super().__init__() self.name = name def run1(self): print(f'{self.name} is running') time.sleep(random.randint(1, 3)) print(f'{self.name} is gone') if __name__ == '__main__': p = MyProcess('常鑫') p.start() print('==主')简单应用
# 简单应用/ from multiprocessing import Process import time def task(name): print(f'{name} is running') time.sleep(1) print(f'{name} is gone') def task1(name): print(f'{name} is running') time.sleep(2) print(f'{name} is gone') def task2(name): print(f'{name} is running') time.sleep(3) print(f'{name} is gone') if __name__ == '__main__': p1 = Process(target=task, args=('常鑫一号英雄',)) p2 = Process(target=task, args=('常鑫二号英雄',)) start_time = time.time() task(1) task1(2) task2(3) print(f'结束时间{time.time() - start_time}') # 三个进程并发或者并行的执行三个任务 # 创建进程是并行,不创建是串行
2.获取进程pid
import os
print(f'子进程:{os.getpid()}')
print(f'主进程:{os.getppid()}')
cmd命令查看pid
tasklist 查看所有进程的pid
tasklist|findstr pycharm 查看pycharm的pidfrom multiprocessing import Process
import os
print(f'子进程:{os.getpid()}')
print(f'主进程:{os.getppid()}')
def task(name):
print(f'子进程:{os.getpid()}')
print(f'主进程:{os.getppid()}')
if __name__ == '__main__':
p = Process(target=task, args=('常鑫',))
p.start()
print('==主开始')
print(f'==主{os.getpid()}')
print(f'===主{os.getppid()}')3.验证进程之间的空间隔离
初始子进程的时候copy主进程,之后主进程和子进程没有任何联系,不共同享用任何内容from multiprocessing import Process
import time
name = '常鑫'
def task():
global name
name = '郭记'
print(f'子进程的名字是: {name}')
if __name__ == '__main__':
p = Process(target=task)
p.start()
time.sleep(1)
print(f'主进程的名字是: {name}')
----------------------------分割线-----------------------------
lst = ['郭苏慧', ]
def task1():
lst.append('郭记')
print(f'子进程的名字是: {lst}')
if __name__ == '__main__':
p = Process(target=task1)
p.start()
time.sleep(2)
print(f'主进程的名字是: {lst}')4. join
join 让主进程等待子进程结束之后再执行主进程
join 只针对主进程,如果join下面多次join 他是不阻塞的
join 就是阻塞,主进程有join,主进程下面的代码一律不执行,直到进程执行完毕之后,在执行.# 正确 重点
from multiprocessing import Process
import time
def task(name):
print(f'{name} is running')
time.sleep(2)
print(f'{name} is gone')
if __name__ == '__main__':
start_time = time.time()
l1 = []
for i in range(1, 4):
p = Process(target=task, args=(i,))
l1.append(p)
p.start()
for i in l1:
i.join()
print(f'==主{time.time() - start_time}')
错误示范:
for i in range(1,4):
p = Process(target=task,args=(i,))
p.start()
p.join()
'''
p1 = Process(target=task,args=(1,))
p1.start()
p1.join()
p2 = Process(target=task,args=(2,))
p2.start()
p2.join()
p3 = Process(target=task,args=(3,))
p3.start()
p3.join()
'''# join让主进程等待子进程结束之后再执行主进程
from multiprocessing import Process
import time
def task(name):
print(f'{name} is running')
time.sleep(2)
print(f'{name} is gone')
if __name__ == '__main__':
p = Process(target=task, args=('常鑫',))
p.start()
p.join()
print('==主进程开始')
# 多个进程使用join
def task(name, sec):
print(f'{name} is running')
time.sleep(sec)
print(f'{name} is gone')
if __name__ == '__main__':
star_time = time.time()
start_time = time.time()
p1 = Process(target=task, args=('常鑫', 1))
p2 = Process(target=task, args=('李业', 2))
p3 = Process(target=task, args=('海狗', 3))
p1.start()
p2.start()
p3.start()
# join 只针对主进程,如果join下面多次join 他是不阻塞的.
p1.join()
p2.join()
p3.join()
print(f'==主{time.time()-start_time}')
# ----------------------------------------------------------
def task(name, sec):
print(f'{name}is running')
time.sleep(sec)
print(f'{name} is gone')
if __name__ == '__main__':
start_time = time.time()
p1 = Process(target=task, args=('常鑫', 3))
p2 = Process(target=task, args=('李业', 2))
p3 = Process(target=task, args=('海狗', 1))
p1.start()
p2.start()
p3.start()
# join就是阻塞
p1.join() # 等2s
print(f'==主1:{time.time() - start_time}')
p2.join()
print(f'===主2:{time.time() - start_time}')
p3.join()
print(f'==主3:{time.time() - start_time}')5.进程的其他参数
p.terminate() # 杀死子进程 ***
print(p.is_alive()) # *** 判断子进程 False True
p.join() # ***from multiprocessing import Process
import time
def task(name):
print(f'{name} is running')
time.sleep(2)
print(f'{name} is gone')
if __name__ == '__main__':
p = Process(target=task, args=('常鑫', ), name='Alex')
p.start()
time.sleep(1)
p.terminate() # 杀死子进程 ***
p.join() # ***
time.sleep(1)
print(p.is_alive()) # *** 判断子进程 False
print(p.name)
p.name = 'sb'
print(p.name)
print('主程序开始')6.守护进程
p.daemon = True
将p子进程设置成守护进程,只要主进程结束,守护进程马上结束 一定要在子进程开启之前设置from multiprocessing import Process
import time
def task(name):
print(f'{name} is running')
time.sleep(2)
print(f'{name} is gone')
if __name__ == '__main__':
p = Process(target=task, args=('常鑫',)) # 创建一个进程对象
p.daemon = True # 将p子进程设置成守护进程,只要主进程结束,守护进程马上结束.
p.start()
# p.daemon = True # 一定要在子进程开启之前设置
time.sleep(1)
print('===主')7.僵尸进程孤儿进程
基于unix环境(linux, macOS)
主进程需要等待子进程结束之后,主进程才结束
==主进程时刻监测子进程的运行状态,当子进程结束之后,一段时间之内,将子进程进行回收.==
为什么主进程不在子进程结束后马上对其回收呢?
- 主进程与子进程是异步关系.主进程无法马上捕获子进程什么时候结束.
- 如果子进程结束之后马上再内存中释放资源,主进程就没有办法监测子进程的状态了.
unix针对于上面的问题,提供了一个机制.所有的子进程结束之后,立马会释放掉文件的操作链接,内存的大部分数据,但是会保留一些内容: ==进程号,结束时间,运行状态==,等待主进程监测,回收.
僵尸进程:==所有的子进程结束之后,在被主进程回收之前,都会进入僵尸进程状态.==
僵尸进程有无危害???
如果父进程不对僵尸进程进行回收(
wait/waitpid),产生大量的僵尸进程,这样就会占用内存,占用进程pid号.孤儿进程:
父进程由于某种原因结束了,但是你的子进程还在运行中,这样你的这些子进程就成了孤儿进程.你的父进程如果结束了,你的所有的孤儿进程就会被
init进程的回收,init就变成了你的父进程,对你进行回收.僵尸进程如何解决???
父进程产生了大量子进程,但是不回收,这样就会形成大量的僵尸进程,解决方式==就是直接杀死父进程==,将所有的僵尸进程变成孤儿进程进程,由
init进行回收.
8.互斥锁
互斥锁:
指散布在不同进程之间的若干程序片断，当某个进程运行其中一个程序片段时，其它进程就不能运行它们之中的任一程序片段，只能等到该进程运行完这个程序片段后才可以运行的一种类似于"锁"的机制
版本一:
现在是所有的进程都并发的抢占打印机.
并发是以效率优先的,但是目前我们的需求: 顺序优先.
多个进程共强一个资源时, 要保证顺序优先: 串行,一个一个来.
版本二:
我们利用join 解决串行的问题,保证了顺序优先,但是这个谁先谁后是固定的.
这样不合理. 你在争抢同一个资源的时候,应该是先到先得,保证公平.
lock与join的区别.
共同点: 都可以把并发变成串行, 保证了顺序.
不同点: join人为设定顺序,lock让其争抢顺序,保证了公平性.版本三:(for i in 循环)
from multiprocessing import Process
from multiprocessing import Lock
import time
import random
import sys
def task1(p, lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1, 3))
print(f'{p}开始打印了')
lock.release()
def task2(p, lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1, 3))
print(f'{p}开始打印了')
lock.release()
def task3(p, lock):
lock.acquire()
print(f'{p}开始打印了')
time.sleep(random.randint(1, 3))
print(f'{p}开始打印了')
lock.release()
if __name__ == '__main__':
mutex = Lock()
for i in range(1, 4):
p = Process(target=getattr(sys.modules[__name__], f'task{i}'), args=(i, mutex))
p.start()9.进程之间的通信
进程在内存级别是隔离的,但是文件在磁盘上
1.基于文件通信
抢票系统.
1. 先可以查票.查询余票数. 并发
2. 进行购买,向服务端发送请求,服务端接收请求,在后端将票数-1,返回到前端. [串行].
当很多进程共强一个资源(数据)时, 你要保证顺序(数据的安全),一定要串行.
互斥锁: 可以公平性的保证顺序以及数据的安全.
基于文件的进程之间的通信:
1.效率低.
2.自己加锁麻烦而且很容易出现死锁.from multiprocessing import Process
from multiprocessing import Lock
import random
import time
import json
import os
def search():
time.sleep(random.randint(1, 3)) # 模拟网络延迟(查询环节)
with open('db.json', encoding='utf-8') as f1:
dic = json.load(f1)
print(f'{os.getpid()}查看了票的剩余量,还剩余{dic["count"]}')
def paid():
with open('db.json', encoding='utf-8') as f1:
dic = json.load(f1)
if dic['count'] > 0:
dic['count'] -= 1
time.sleep(random.randint(1, 3))
with open('db.json', encoding='utf-8', mode='w') as f1:
json.dump(dic, f1)
print(f'{os.getpid()}购票成功, 还剩余{dic["count"]}票')
else:
time.sleep(1)
print(f'{os.getpid()}购票未成功')
def task(lock):
search()
lock.acquire()
paid()
lock.release()
if __name__ == '__main__':
mutex = Lock()
for i in range(5):
p = Process(target=task, args=(mutex,))
p.start()2.基于队列通信
队列: 把队列理解成一个容器,这个容器可以承载一些数据,
队列的特性: 先进先出永远保持这个数据. FIFO 羽毛球筒.
q.put(5555) 当队列满了时,在进程put数据就会阻塞.
print(q.get()) 当数据取完时,在进程get数据也会出现阻塞,直到某一个进程put数据.
print(q.get(timeout=3)) 阻塞3秒,3秒之后还阻塞直接报错.
print(q.get(block=False)) 只要遇到阻塞就会报错.利用队列Queue改进选票系统作业:
- 票数放入队列中存储
- 开启多个进程进行选票，查票为并发效果，买票为串行效果
- 购买成功、失败都需要提示
from multiprocessing import Process
from multiprocessing import Queue
import random
import time
import os
def search(q):
get = q.get() # 取出
print(f'{os.getpid()}查看了票的剩余量,还剩余{get["count"]}')
q.put(get) # 输入
def paid(q):
time.sleep(random.randint(1, 3))
q_dic = q.get() # 取出
q.put(q_dic) # 输入
if q_dic["count"] > 0:
q_dic["count"] -= 1
print(f"{os.getpid()}购买成功!{q_dic['count']} ")
try:
q.put(q_dic, block=False)
except Exception:
pass
else:
print(f"{os.getpid()}购买失败")
def task(q):
search(q)
paid(q)
if __name__ == '__main__':
q = Queue(1)
q.put({"count": 3})
for i in range(5):
p = Process(target=task, args=(q,))
p.start()模拟双十一排队抢小米手机，多用户抢购，只能选取前10个用户:
开启多个用户抢购买手机。
只能限定10人购买。
最终将10个用户的排名展示出来。
import os
from multiprocessing import Queue
from multiprocessing import Process
def task(q):
try:
q.put(f'{os.getpid()}', block=False)
except Exception:
return
if __name__ == '__main__':
q = Queue(10)
for i in range(100):
p = Process(target=task, args=(q,))
p.start()
for i in range(1, 10):
print(f'排名第{i}的用户是{q.get()}')10.生产者消费者模块
编程思想,模型,设计模式,理论等等,都是交给你一种编程的方法,以后你遇到类似的情况,套用即可.
生产者消费者模型三要素:
生产者: 产生数据的
消费者: 接收数据做进一步处理的
容器: 盆(队列)
那么队列容器起到什么作用? 起到缓冲的作用,平衡生产力与消费力,解耦.
from multiprocessing import Process
from multiprocessing import Queue
import random
import time
def producer(q, name):
for i in range(1, 6):
time.sleep(random.randint(1, 2))
res = f'{i}号包子'
q.put(res)
print(f'生产者{name}生产了{res}')
def consumer(q, name):
while 1:
try:
food = q.get(timeout=3)
time.sleep(random.randint(1, 3))
print(f'消费者{name}吃了吃了吃了{food}')
except Exception:
pass
if __name__ == '__main__':
q = Queue()
p1 = Process(target=producer, args=(q, '常鑫'))
p2 = Process(target=consumer, args=(q, '常鑫鑫'))
p1.start()
p2.start()2. thread
1.线程的理论知识
1. 什么是线程:进程是资源单位, 线程是执行单位.
进程:进程会在内存中开辟一个进程空间,将主进程的资料数据全部复制一份,线程会执行里面的代码.
2. 线程vs进程:
1. 开启进程的开销非常大,比开启线程的开销大很多.
2. 开启线程的速度非常快,要快几十倍到上百倍.
3. 线程与线程之间可以共享数据,进程与进程之间需借助队列等方法实现通信.
3. 线程的应用: 数据共享, 开销小,速度快,
并发: 一个cpu 看起来像是同时执行多个任务,单个进程开启三个线程,并发的执行任务
主线程子线程没有地位之分,但是,一个进程谁在干活? 一个主线程在干活,当干完活了,你得等待其他线程干完活之后,才能结束本进程==什么是线程==
一条流水线的工作流程.
进程: 在内存中开启一个进程空间,然后将主进程的所有的资源数据复制一份,然后调用cpu去执行这些代码.
之前的描述不够具体:
开启一个进程:
在内存中开启一个进程空间,然后将主进程的所有的资源数据复制一份,然后调用线程去执行代码
进程是资源单位, 线程是执行单位.
以后你描述开启一个进程:
开启一个进程:进程会在内存中开辟一个进程空间,将主进程的资料数据全部复制一份,线程会执行里面的代码.
==线程vs进程==
- 开启进程的开销非常大,比开启线程的开销大很多.
- 开启线程的速度非常快.要快几十倍到上百倍.
- 线程线程之间可以共享数据,进程与进程之间需借助队列等方法实现通信.
==线程的应用==
并发: 一个cpu 看起来像是同时执行多个任务.
单个进程开启三个线程.并发的执行任务.
开启三个进程并发的执行任务.
文本编辑器:
- 输入文字.
- 在屏幕上显示.
- 保存在磁盘中.
开启多线程就非常好了:
数据共享, 开销小,速度快.
主线程子线程没有地位之分,但是,一个进程谁在干活? 一个主线程在干活,当干完活了,你得等待其他线程干完活之后,才能结束本进程
2.开启线程的两张方式
第一种方式:
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(1)
print(f'{name} in gone')
if __name__ == '__main__':
p1 = Thread(target=task, args=('常鑫',))
p1.start()
print('===主线程')第二种方式:
from threading import Thread
import time
class MyThread(Thread):
def __init__(self, name, l1, s1):
super().__init__()
self.name = name
self.l1 = l1
self.s1 = s1
def run(self):
print(f'{self.name} is running')
print(f'{self.l1} is running')
print(f'{self.s1} is running')
time.sleep(1)
print(f'{self.name} is gone')
print(f'{self.l1} is gone')
print(f'{self.s1} is gone')
if __name__ == '__main__':
p1 = MyThread('常鑫', [1, 2, 3], '100')
p1.start()
print('===主线程')3.线程vs进程的代码对比
开启速度对比,线程比进程
from multiprocessing import Process def work(): print('hello') if __name__ == '__main__': # 在主进程下开启线程 t = Process(target=work) t.start() print('主线程/主进程')from threading import Thread import time def task(name): print(f'{name} is running') time.sleep(1) print(f'{name} is gone') if __name__ == '__main__': t1 = Thread(target=task, args=('海狗',)) t1.start() print('===主线程') # 线程是没有主次之分的.对比
pid==同一个pid==from threading import Thread import os def task(): print(os.getpid()) if __name__ == '__main__': t1 = Thread(target=task) t2 = Thread(target=task) t1.start() t2.start() print(f'===主线程{os.getpid()}')同一个进程内线程共享内部数据
同一进程内的资源数据对于这个进程的多个线程来说是共享的. from threading import Thread x = 3 def task(): global x x = 100 if __name__ == '__main__': t1 = Thread(target=task) t1.start() t1.join() print(f'===主线程{x}')
4.线程的相关其他方法(了解)
# Thread实例对象的方法
p1.setName('子线程1') # 设置线程名
p1.getName() # 返回线程名
---print(p1.name) # 获取线程名 ***
print(p1.isAlive()) # 返回线程是否活动的。
# threading模块提供的一些方法：
print(current_thread()) # 获取当前线程的对象
print(currentThread()) # 获取当前线程的对象
print(enumerate()) # 返回一个列表,包含所有的线程对象
---print(activeCount()) # *** 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。from threading import Thread
from threading import currentThread
from threading import enumerate
from threading import activeCount
import os
import time
x = 9
def task():
print(currentThread())
time.sleep(1)
print('666')
if __name__ == '__main__':
p1 = Thread(target=task, name='p1') # name 设置线程名
p2 = Thread(target=task, name='p2') # name 设置线程名
p1.start()
p2.start()
# Thread实例对象的方法
p1.setName('子线程1') # 设置线程名
p2.setName('子线程1') # 设置线程名
p1.getName() # 返回线程名
p2.getName() # 返回线程名
print(p1.name) # 获取线程名 ***
print(p2.name) # 获取线程名 ***
print(p1.isAlive()) # 返回线程是否活动的。
print(p2.isAlive()) # 返回线程是否活动的。
# threading模块提供的一些方法：
print(currentThread()) # 获取当前线程的对象
print(enumerate()) # 返回一个列表,包含所有的线程对象
print(activeCount()) # *** 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
print(f'主线程{os.getpid()}')5.守护线程(考点)
join: 阻塞 告知主线程要等待我子线程执行完毕之后再执行主线程
主线程什么时候结束???
守护线程 等待非守护子线程以及主线程结束之后,结束.from threading import Thread
import time
def foo():
print(123) # 1
time.sleep(1)
print("end123") # 4
def bar():
print(456) # 2
time.sleep(2)
print("end456") # 5
t1 = Thread(target=foo)
t2 = Thread(target=bar)
t1.daemon = True
t1.start()
t2.start()
print("main-------") # 3
# 结果:
# 123
# 456
# main-------
# end123
# end4566.互斥锁(考点)
正常情况加锁之后编程串行
锁之后加上延迟就不一定,有的可能就会出现插队现象from threading import Thread
from threading import Lock
import time
import random
x = 10
def task(lock):
lock.acquire()
time.sleep(random.randint(1, 3)) # 卡点
global x
temp = x
time.sleep(0.1)
temp = temp - 1
x = temp
lock.release()
if __name__ == '__main__':
mutex = Lock()
l1 = []
for i in range(10):
t = Thread(target=task, args=(mutex,))
l1.append(t)
t.start()
time.sleep(1)
print(f'主线程{x}')7.死锁现象与递归锁
- 死锁现象就是: A进程那着A钥匙去找B钥匙,B进程拿着B钥匙去找A钥匙
- 递归锁: 可以解决死锁现象,业务需要多个锁时,优先考虑递归锁
- 锁必须写成==
lock_A = lock_B = RLock()==格式,原理是==pid==都一样,每锁一次,锁的数量加一,解开的时候减一.锁的数量如果不为零其他线程不能抢锁==from threading import RLock==导入模块
死锁现象:
from threading import Thread
from threading import Lock
import time
lock_A = Lock()
lock_B = Lock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
lock_A.acquire()
print(f'{self.name}拿到的A')
lock_B.acquire()
print(f'{self.name}拿到的B')
lock_B.release()
lock_A.release()
def f2(self):
lock_B.acquire()
print(f'{self.name}拿到的B')
time.sleep(0.1)
lock_A.acquire()
print(f'{self.name}拿到的A')
lock_A.release()
lock_B.release()
if __name__ == '__main__':
for i in range(3):
t = MyThread()
t.start()递归锁:
from threading import Thread
from threading import RLock
import time
lock_A = lock_B = RLock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
lock_A.acquire()
print(f'{self.name}拿到的A')
lock_B.acquire()
print(f'{self.name}拿到的B')
lock_B.release()
lock_A.release()
def f2(self):
lock_B.acquire()
print(f'{self.name}拿到的B')
lock_A.acquire()
print(f'{self.name}拿到的A')
time.sleep(1)
lock_A.release()
lock_B.release()
if __name__ == '__main__':
for i in range(3):
t = MyThread()
t.start()8.信号量
也是一种锁, 控制并发数量
==from threading import current_thread==获取当前线程的对象模块
==from threading import Semaphore==导入信号量模块
==sem = Semaphore(5)==实例化信号量 不写时无穷大
==sem.acquire()==函数里面获取信号量
from threading import Thread
from threading import Semaphore
from threading import current_thread
import random
import time
sem = Semaphore(5)
def task():
sem.acquire()
print(f'{current_thread().name} 房间')
time.sleep(random.randint(1, 3))
sem.release()
if __name__ == '__main__':
for i in range(30):
t = Thread(target=task, )
t.start()9. ==GIL==全局解释器锁
好多自称大神的说,GIL锁就是python的致命缺陷,Python不能多核,并发不行等等 …..
==理论上来说:单个进程的多线程可以利用多核.==
但是,开发
Cpython解释器的程序员,给进入解释器的线程加了锁.为什么加锁?
- 当时都是单核时代,而且
cpu价格非常贵. - 如果不加全局解释器锁, 开发
Cpython解释器的程序员就会在源码内部各种主动加锁,解锁,非常麻烦,各种死锁现象等等.他为了省事儿,直接进入解释器时给线程加一个锁. - ==优点: 保证了
Cpython解释器的数据资源的安全.== - ==缺点: 单个进程的多线程不能利用多核.==
- 当时都是单核时代,而且
Jpython没有GIL锁.pypy也没有GIL锁.现在多核时代, 我将
Cpython的GIL锁去掉行么?因为
Cpython解释器所有的业务逻辑都是围绕着单个线程实现的,去掉这个GIL锁,几乎不可能.==单个进程的多线程可以并发,但是不能利用多核,不能并行. 多个进程可以并发,并行.==
==io密集型: 单个进程的多线程合适,并发执行==
==计算密集型:多进程的并行==
10.==GIL==锁与==lock==锁的区别
- 相同点: 都是同种锁,互斥锁.
- 不同点:
- GIL锁全局解释器锁,保护解释器内部的资源数据的安全.
- GIL锁 上锁,释放无需手动操作.
- 自己代码中定义的互斥锁保护进程中的资源数据的安全.
- 自己定义的互斥锁必须自己手动上锁,释放锁.
11.验证计算密集型IO密集型的效率
==
io密集型: 单个进程的多线程的并发效率高合适.并发执行====计算密集型:多进程的并发并行效率高.并行==
代码验证:
计算密集型: 单个进程的多线程并发 vs 多个进程的并发并行 from multiprocessing import Process from threading import Thread import time def task(): count = 0 for i in range(30000000): # (三千万) count += 1 if __name__ == '__main__': # 多进程的并发,并行 2.3737263679504395秒 start_time = time.time() l1 = [] for i in range(4): p = Process(target=task,) l1.append(p) p.start() for i in l1: i.join() print(f'执行时间:{time.time()-start_time}') # 多线程的并发 6.290118932723999秒 start_time = time.time() l1 = [] for i in range(4): p = Thread(target=task,) l1.append(p) p.start() for i in l1: i.join() print(f'执行时间:{time.time()-start_time}') 计算密集型: 多进程的并发并行效率高.# IO密集型: 单个进程的多线程并发 vs 多个进程的并发并行 from multiprocessing import Process from threading import Thread import time def task(): count = 0 time.sleep(1) count += 1 if __name__ == '__main__': # 多进程的并发,并行 3.0123958587646484秒 start_time = time.time() l1 = [] for i in range(50): p = Process(target=task, ) l1.append(p) p.start() for p in l1: p.join() print(f'执行效率:{time.time() - start_time}') # 多线程的并发 1.0087950229644775秒 start_time = time.time() l1 = [] for i in range(50): p = Thread(target=task,) l1.append(p) p.start() for p in l1: p.join() print(f'执行效率:{time.time()- start_time}') 对于IO密集型: 单个进程的多线程的并发效率高.
12.多线程实现==socket==通信
无论是多线程还是多进程,如果按照之前的写法,来一个客户端请求,我就开一个线程,来一个请求开一个线程,应该是这样: 你的计算机允许范围内,开启的线程进程数量越多越好.
服务端:
from threading import Thread
import socket
def communicate(conn, addr):
while 1:
try:
from_client_data = conn.recv(1024)
print(f'来{addr[1]}的信息{from_client_data.decode("utf-8")}')
to_client_data = input('>>>').strip()
conn.send(to_client_data.encode('utf-8'))
except Exception:
break
conn.close()
def _accket():
server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen(5)
while 1:
conn, addr = server.accept()
t = Thread(target=communicate, args=(conn, addr))
t.start()
if __name__ == '__main__':
_accket()客户端:
import socket
client = socket.socket()
client.connect(('127.0.0.1', 8080))
while 1:
try:
to_server_data = input('>>>').strip()
client.send(to_server_data.encode('utf-8'))
from_server_data = client.recv(1024)
print(f'来自服务器的消息: {from_server_data.decode("utf-8")}')
except Exception:
break
client.close()13.进程池线程池
from concurrent.futures import ProcessPoolExecutor # 线程池模块
from concurrent.futures import ThreadPoolExecutor # 进程池模块
p = ProcessPoolExecutor() # 默认不写,进程池里面的进程数与cpu核个数相等(并行(并行+并发))
t = ThreadPoolExecutor() # 默认不写, cpu核个数*5 线程数 (并发)
print(os.cpu_count()) # 查看电脑几核from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
import random
import time
import os
print(os.cpu_count()) # 查看电脑几核
def task():
print(f'pid号: {os.getpid()} 来了')
time.sleep(random.randint(1, 3))
if __name__ == '__main__':
# 开启进程池 (并行(并行+并发))
p = ProcessPoolExecutor() # 默认不写,进程池里面的进程数与cpu个数相等
for i in range(20):
p.submit(task, )
# 开启线程池 (并发)
t = ThreadPoolExecutor() # 默认不写, cpu个数*5 线程数
for i in range(40):
t.submit(task, )14.阻塞 非阻塞 异步 同步
问题出在哪里?
1. 分析结果的过程是串行,效率低.
2. 你将所有的结果全部都爬取成功之后,放在一个列表中,分析.
问题1解决:
在开进程池,再开进程,耗费资源.
'''
爬取一个网页需要2s,并发爬取10个网页:2.多s.
分析任务: 1s. 10s. 总共12.多秒.
现在这个版本的过程:
异步发出10个爬取网页的任务,然后4个进程并发(并行)的先去完成4个爬取网页的任务,然后谁先结束,谁进行下一个
爬取任务,直至10个任务全部爬取成功.
将10个爬取结果放在一个列表中,串行的分析.
爬取一个网页需要2s,分析任务: 1s,总共3s,总共3.多秒(开启进程损耗).
. 10s.
下一个版本的过程:
异步发出10个 爬取网页+分析 的任务,然后4个进程并发(并行)的先去完成4个爬取网页+分析 的任务,
然后谁先结束,谁进行下一个 爬取+分析 任务,直至10个爬取+分析 任务全部完成成功.
'''
回调函数是主进程帮助你实现的, 回调函数帮你进行分析任务. 明确了进程的任务: 只有一个网络爬取.
分析任务: 回调函数执行了.对函数之间解耦.
极值情况: 如果回调函数是IO任务,那么由于你的回调函数是主进程做的,所以有可能影响效率.
回调不是万能的,如果回调的任务是IO,
那么异步 + 回调机制 不好.此时如果你要效率只能牺牲开销,再开一个线程进程池.
异步就是回调! 这个是错的!! 异步,回调是两个概念.
'''
如果多个任务,多进程多线程处理的IO任务.
# 1. 剩下的任务 非IO阻塞. 异步 + 回调机制
# 2. 剩下的任务 IO << 多个任务的IO 异步 + 回调机制
# 3. 剩下的任务 IO >= 多个任务的IO 第二种解决方式,或者两个进程线程池.
'''15.异步 调用机制
16.事件==Event==
17.协程的初识
18.协程
