目录Python基础之网络并发编程
一、任务运行的多种方式
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()