一、任务运行的多种方式
1、同步与异步
根据进程和函数之间的通讯机制,函数可分为同步和异步
同步与异步是为了表达任务的提交方式
-
同步:指提交任务后,直接得到任务的最终结果
- eg:在取款机前排队取款,直到排队取到钱之后离开
-
异步:指提交任务后,不等待任务的结果,直接去做另外的事情,任务完成后结果会反馈回来
- eg:指在银行取钱,领到取票机的小票后,不用一直排队,可以去做其他的事情,到号时,银行会叫号
2、阻塞与非阻塞
指函数或方法调用的时候,结果是否会立即返回
- 阻塞:执行任务后会立即返回任务的结果
- 非阻塞:执行任务后,并不会立即返回结果,就绪态、运行态
3、综合使用
综合使用指同步与非同步、阻塞与非阻塞四种状态两两结合使用
-
同步阻塞
- 指调用任务后,遇到阻塞,但是过程中什么也干不了,只能等待任务的最终结果
-
同步非阻塞
- 指调用任务后,任务在顺利进行,运行通畅,直到获取任务结果
-
异步阻塞
- 指调用任务后,任务遇到阻塞,没有办法直接获取到最终结果,但是调用者可以去处理其他的事情,并不会影响调用者处理其他事情,等原任务执行结束后反馈结果即可
-
异步非阻塞
- 指调用者提交一个任务后,立即提交其他的任务,并且提交的多个任务都可以通畅运行,立即获得结果,这种状态的效率极高
二、创建进程的多种方式
通常,用户使用计算机,创建进程都是通过双击电脑桌面上的应用图标来创建进程,但我们程序员在使用python时,可以直接通过代码来创建进程,并且可以同时创建多个进程,且进程与进程之间互不干扰,可以高效的提高任务完成率
1、双击电脑桌面图标
2、通过代码创建
须知:
-
windows
- 以导入模块的方式创建进程
-
linux/mac
- 以拷贝代码的方式创建进程
1、代码创建进程
方式一
函数方式创建
from multiprocessing import Process
import time
def task(name):
print('task is running',name)
time.sleep(3)
print('task is over',name)
if __name__ == '__main__':
# p1 = Process(target=task, args=('jason',)) # 位置参数
p1 = Process(target=task, kwargs={'name':'jason123'}) # 关键字参数
p1.start() # 异步 告诉操作系统创建一个新的进程 并在该进程中执行task函数
task() # 同步
print('主')
方式二
继承类的方式创建
from multiprocessing import Process
import time
class MyProcess(Process):
def __init__(self, name, age):
super().__init__()
self.name = name
self.age = age
def run(self):
print('run is running', self.name, self.age)
time.sleep(3)
print('run is over', self.name, self.age)
if __name__ == '__main__':
obj = MyProcess('jason', 123)
obj.start()
print('主')
2、进程间数据隔离
进程隔离是指在同一台计算机上多个同时运行的进程,进程与进程之间数据默认是隔离的
eg:在主进程定义一个变量,使用主进程创建多个进程,在创建的多个进程内修改、重新定义主进程的变量,每个进程内的变量互不干扰,包括主进程。
代码表现:
from multiprocessing import Process
import time
money = 1000
def task():
global money
money = 666
print('子进程的task函数查看money', money)
if __name__ == '__main__':
p1 = Process(target=task)
p1.start() # 创建子进程
time.sleep(3) # 主进程代码等待3秒
print(money) # 主进程代码打印money
3、进程join方法
使用主进程创建多个子进程后,代码运行到主进程创建的子进程代码后,通常子进程代码是异步执行,并不会干扰主进程代码执行顺序,通过join方法可使异步执行的代码变为同步执行,主进程会等待子进程代码运行结束后向下执行,join方法需定义在子进程启动代码后,否则会影响同步执行的结果
使用方法:
通过定义的子进程对象‘点’join()的方式使用
代码表现:
from multiprocessing import Process
import time
def task(name, n):
print('%s is running' % name)
time.sleep(n)
print('%s is over' % name)
if __name__ == '__main__':
p1 = Process(target=task, args=('jason1', 1))
p2 = Process(target=task, args=('jason2', 2))
p3 = Process(target=task, args=('jason3', 3))
# p.start() # 异步
'''主进程代码等待子进程代码运行结束再执行'''
# p.join()
# print('主')
start_time = time.time()
p1.start()
p1.join()
p2.start()
p2.join()
p3.start()
p3.join()
# p1.join()
# p2.join()
# p3.join()
print(time.time() - start_time) # 3秒多
4、IPC机制
IPC>>>:进程间通讯机制
相当于消息队列,储存数据的地方,同一主进程下的所有子进程共同存、取数据的地方,大家都可以使用
使用方法:
1、导入multiprocessing模块下的Queue
2、通过赋值变量的方式创建消息队列
3、消息队列后方括号内为储存数据的个数
4、对象‘点’put():存入
5、对象‘点’get():取
代码用法:
from multiprocessing import Queue
q = Queue(3) # 括号内可以指定存储数据的个数
# 往消息队列中存放数据
q.put(111)
# print(q.full()) # 判断队列是否已满
q.put(222)
q.put(333)
# print(q.full()) # 判断队列是否已满
# 从消息队列中取出数据
print(q.get())
print(q.get())
# print(q.empty()) # 判断队列是否为空
print(q.get())
# print(q.empty()) # 判断队列是否为空
# print(q.get())
print(q.get_nowait())
"""
full() empty() 在多进程中都不能使用!!!
"""
from multiprocessing import Process, Queue
def product(q):
q.put('子进程p添加的数据')
def consumer(q):
print('子进程获取队列中的数据', q.get())
if __name__ == '__main__':
q = Queue()
# 主进程往队列中添加数据
# q.put('我是主进程添加的数据')
p1 = Process(target=consumer, args=(q,))
p2 = Process(target=product, args=(q,))
p1.start()
p2.start()
print('主')
三、生产者消费者模型
-
生产者:
-
负责生产数据的代码
eg: 回想使用爬虫,在网页上获取数据
-
-
消费者:
- 负责处理数据的代码
eg:类似于正则,对数据进行处理,分类
该模型除了有生产者和消费者之外还必须有消息队列(只要是能够提供数据保存服务和提取服务的理论上都可以)
四、进程对象的多种方法
1、查看进程号
方式一:
1、导入multiprocessing模块下current_process方法
2、在进程下使用currnet_process().pid
3、该方法可直接查看进程的 '父' 进程,以及进程号
方式二:
1、导入os模块
2、os.getpid()
查看当前进程的进程号
3、os.getppid()
查看进程的 '父' 进程
2、终止进程
1、通过进程对象名‘点’terminate()
可直接结束进程
ps: 计算机操作系统有很多命令可以直接杀死(结束)进程
3、判断进程是否存活
1、通过进程名‘点’is_alive()
存活:Trun
死亡:False
4、守护进程
守护进程是主进程创建的进程
- 守护进程会在主进程代码运行结束后就立即终止
- 守护进程内无法创建子进程,否则报错
- 守护进程和主进程之间是相互独立的
代码用法:
from multiprocessing import Process
import time
def task(name):
print('德邦总管:%s' % name)
time.sleep(3)
print('德邦总管:%s' % name)
if __name__ == '__main__':
p1 = Process(target=task, args=('大张红',))
p1.daemon = True
p1.start()
time.sleep(1)
print('恕瑞玛皇帝:小吴勇嗝屁了')
5、僵死进程与孤儿进程
5、1.僵尸进程
僵尸进程是指进程运行结束后,并不会立刻销毁所有的数据,而是等一段时间后才会销毁,
eg:进程号、进程执行时间、进程执行结果,会将这些结果反馈给父进程查看
所有的进程都会经理僵尸进程这个阶段
5、2.孤儿进程
指子进程正常运行,父进程以外死亡,操作系统会针对孤儿进程派出福利院进行接收、接管这些进程
福利院:__ init __进程/进程号为1
福利院会对孤儿进程进行接管、负责收集孤儿进程的数据、结果、执行时间
五、互斥锁
1、多进程错乱问题
多进程错乱问题指,当我们使用主进程创建了多个子进程后,这些子进程,在同一时间查询并需要使用主进程内IPC(消息队列)内数据,但是因为是同一时间,所以会产生消息错乱问题
eg:主进程消息队列内只有一个数据,但是所有子进程在同一时间调用get方法后都可以查询并取到数据
解决办法:可以使用互斥锁,将并发变为穿行,牺牲了效率,保证了数据的准确性和安全
多进程错乱问题代码表现:
模拟抢票软件
from multiprocessing import Process
import time
import json
import random
# 查票
def search(name):
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
print('%s在查票 当前余票为:%s' % (name, data.get('ticket_num')))
# 买票
def buy(name):
# 再次确认票
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
# 模拟网络延迟
time.sleep(random.randint(1, 3))
# 判断是否有票 有就买
if data.get('ticket_num') > 0:
data['ticket_num'] -= 1
with open(r'data.json', 'w', encoding='utf8') as f:
json.dump(data, f)
print('%s买票成功' % name)
else:
print('%s很倒霉 没有抢到票' % name)
def run(name):
search(name)
buy(name)
if __name__ == '__main__':
for i in range(10):
p = Process(target=run, args=('用户%s'%i, ))
p.start()
2、互斥锁代码实操
互斥锁可以解决多进程错乱问题
用法:
1、在主进程生成互斥锁对象
2、在子进程需要加互斥锁的代码上下方加入关键词
3、只有第一个抢到互斥锁的进程对象才有资格运行代码,在代码运行结束后下一个进程按照上述流程运行
注意事项:
建议将锁加在实际数据操作部分,否则会降低程序运行效率
关键词 | 作用 |
---|---|
Lock | 模块名 |
acquire | 抢锁 |
release | 释放锁 |
代码实操:
from multiprocessing import Process, Lock
import time
import json
import random
def search(name):
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
print('%s查看票 目前剩余:%s' % (name, data.get('ticket_num')))
def buy(name):
# 先查询票数
with open(r'data.json', 'r', encoding='utf8') as f:
data = json.load(f)
# 模拟网络延迟
time.sleep(random.randint(1, 3))
# 买票
if data.get('ticket_num') > 0:
with open(r'data.json', 'w', encoding='utf8') as f:
data['ticket_num'] -= 1
json.dump(data, f)
print('%s 买票成功' % name)
else:
print('%s 买票失败 非常可怜 没车回去了!!!' % name)
def run(name, mutex):
search(name)
mutex.acquire() # 抢锁
buy(name)
mutex.release() # 释放锁
if __name__ == '__main__':
mutex = Lock() # 产生一把锁
for i in range(10):
p = Process(target=run, args=('用户%s号' % i, mutex))
p.start()
六、多进程TCP服务端并发
1、代码用法
import socket
from multiprocessing import Process
def get_server():
server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen(5)
return server
def get_talk(sock):
while True:
data = sock.recv(1024)
print(data.decode('utf8'))
sock.send(data.upper())
if __name__ == '__main__':
server = get_server()
while True:
sock, addr = server.accept()
# 开设多进程去聊天
p = Process(target=get_talk, args=(sock,))
p.start()