网络编程进阶：操作系统介绍、并发编程之多进程

操作系统介绍：

进程：即正在执行的一个过程；进程是对正在运行的程序的一个抽象概念。

操作系统的两大核心功能： 1. 隐藏了硬件的调用接口，为应用程序提供封装好的调用硬件资源的系统调用接口； 2. 将应用程序对硬件资源的竞态请求变得有序化（即管理、调度进程）

多道技术（针对单核，实现并发）：多道指的是多个程序，多道技术的实现是为了解决多个程序竞争或者说共享同一个资源（比如CPU）的有序调度问题，解决方式即多路复用，多路复用分为时间上的复用和空间上的复用

　　空间上的复用：将内存分为若干个部分，每个部分放入一个程序，这样同一时间内存中就有了多道程序；（需要在物理条件上实现不同程序内存地址的互相隔离）

　　时间上的复用：当一个程序在等待I/O时，另一个程序可以使用CPU，如果内存中可以同时存放足够多的程序，则CPU的利用率可以接近100%；（操作系统采用多道技术后，可以控制进程的切换，或者说进程之间去争抢CPU的执行权限。这种切换不仅会在一个进程遇到io时进行，一个进程占用CPU时间过程也会切换，或者说被操作系统夺走CPU的执行权限；但遇到io时的切换会提高CPU的执行效率，占用CPU时间过长的切换不会提高CPU效率，返回会降低效率）；（切换之前将进程的状态保存下来，这样才能保证下次切换回来时，能基于上次切走的位置继续运行）；（由于CPU切换速度特别快，所以给用户的感觉就是看起来是并发）

 

并发编程：

多进程：

进程理论：正在进行的一个过程或者说一个任务。负责执行任务的是CPU

进程和程序的区别：程序金金是一堆代码而已，而进程指的是程序的运行过程

强调：同一个程序执行两次，那也是两个进程

并发和并行： 无论并发还是并行，在用户看来都是“同时”运行的，不管是进程还是线程，都是一个任务而已，真正干活的是CPU，CPU来执行这些任务，而一个CPU同一时刻只能执行一个任务

并发：伪并行，即看起来是同时运行；单个CPU+多道技术就可以实现并发；

并行：真正意义上的同时运行，只有具备多个CPU才能实现并行

进程的创建：但凡是硬件，都需要有操作系统去管理，只要有操作系统，就有进程的概念，就需要创建进程的方式； 创建新进程的形式主要分4种，我需要掌握的是：一个进程在运行过程中开启了子进程（如Nginx开启多进程，os.fork，subprocess.Popen等）

无论哪一种创建进程的方式，新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建的（即已经存在的进程向操作系统发出创建新进程的请求，然后操作系统给这个新创建的进程申请一个新的独立的内存地址）

关于创建的子进程，Unix和Windows：

1. 相同的是： 进程创建后，父进程和子进程有各自不同的内存地址（多道技术要求物理层面实现进程之间内存的隔离），任何一个进程的在其地址空间中的数据修改都不会影响到另外的进程。

2. 不同的是： 在Unix中，子进程的初始地址空间是父进程的一个副本（子进程和父进程是可以有只读的共享内存区的），；但是对于Windows系统来说，从一开始父进程和子进程的地址空间就是不同的

进程的状态：运行、阻塞、就绪

有两种情况会导致一个进程在逻辑上不能运行：

1. 进程挂起是自身原因，遇到I/O阻塞，便让出CPU让其他进程去执行，这样保证CPU一直在工作

2. 与进程无关，是操作系统层面的，可能会因为一个进程占用时间过多，或者优先级等原因，而调用其他的进程去使用CPU

 

开启子进程的两种方式：

方式一：

from multiprocessing import Process
import time
def test(name):
print("%s is running"%name)
time.sleep(3)
print("%s is done"%name)
"""windows系统需要把开启进程的指令放到 if __name__ == "__main__": 下面"""
if __name__ == "__main__":
p = Process(target=test,args=("子进程1",))
# 得到一个对象 # target表示调用对象，即子进程要执行的任务；为函数传参有两种方式：1.按照位置传参： args= 元祖（元祖中的元素会根据列表拆包的方式分配给各个参数值）， 2. 如下
# p = Process(target=test,kwargs={"name":"子进程1"})
# kwargs=字典，字符串格式的形参名作为字典的key
p.start()
# 开启一个子进程，去执行target处的函数
# 应用程序根本开不了子进程；所以 p=start()仅仅是给操作系统发了一个信号
print("主进程")

方式二：

from multiprocessing import Process
import time
class MyProcess(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
# 子进程要执行的任务一定要定义成 run
print("%s is running" % self.name)
time.sleep(3)
print("%s is done"%self.name)
if __name__ == "__main__":
p = MyProcess("子进程1")
p.start()
# 此时的p.start()会自动调用 MyProcess中的 run方法
print("主进程")

 

查看进程的pid和ppid：

from multiprocessing import Process
import time,os
class MyProcess(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
print("%s is running, parent id is <%s>" %(os.getpid(),os.getppid()))
# os.getpid()是拿到当前进程的id， os.getppid()是拿到其父进程的id
# 这行代码中的os.getppid()就是main中的os.getpid()
time.sleep(3)
print("%s is done,parent id is <%s>"%(os.getpid(),os.getppid()))
if __name__ == "__main__":
p = MyProcess("子进程1")
p.start()
print("主进程",os.getpid(),os.getppid())
# 这个里面的os.getpid()是这段代码的进程id；os.getppid()是执行这段代码的进程id(谁来运行这个py文件就是谁)：在pycharm上就是pycharm的pid，在cmd上执行的就是cmd的pid
"""
在cmd中查看某进程的pid：
Windows系统：
tasklist | findstr pycharm
Linux系统：
ps aux |grep pycharm
"""

 

join方法：让主进程卡在原地等子进程执行结束

示例一：

from multiprocessing import Process
import time,os
def task():
print("%s is running,parent id is <%s>"%(os.getpid(),os.getppid()))
time.sleep(3)
print("%s is done,parent id is <%s>"%(os.getpid(),os.getppid()))
if __name__ == "__main__":
p = Process(target=task)
p.start()
p.join()
# 主进程卡在这一步等待子进程执行结束
print("主进程",os.getpid(),os.getppid())
print(p.pid)
#
p.pid 是查询 p的pid
# 子进程结束之后，还会留下僵尸进程；所以子进程执行结束之后，它的pid还存在，即子进程结束之后它的资源并没有被全部回收；但主进程执行完之后，所有的僵尸进程资源都会被回收掉

 

示例二：

from multiprocessing import Process
import time
def task(name,sec):
print("%s is running"%name)
time.sleep(sec)
if __name__ == "__main__":
start = time.time()
p1 = Process(target=task,args=("子进程1",5))
# args=("子进程1",) 一定要加上逗号，这样才是一个元祖
p2 = Process(target=task, args=("子进程2",3))
p3 = Process(target=task, args=("子进程3",2))
p1.start()
p2.start()
p3.start()
# 并不一定是按照p1、p2、p3的顺序执行子进程，因为start（）仅仅是给操作系统发送了一个信号，而操作系统同时在接收很多请求，对于这些请求操作系统是统筹管理的，它先先让谁运行，你并不知道

p1.join()
# 这3个join依然是并行关系，上面3个start发送完信号后，虽然程序卡在了p1.join()这一步，但与此同时，子进程2和子进程3也许已经在执行了
p2.join()
# 主进程卡在p1.join()这一步的时候，有可能子进程2已经执行完毕了，这时主进程就不会卡在p2.join（）这一步了

p3.join()
print("主进程",time.time()-start) 

运行结果之一如下所示：

上述代码中的p1、p2、p3想要实现“串行”，可利用如下方法：

p1.start()
# start（）立马join（）

p1.join()
p2.start()
p2.join()
p3.start()
p3.join()

上述的并发过程还能简写，如下：

if __name__ == "__main__":
start = time.time()
p1 = Process(target=task,args=("子进程1",5))
# args=("子进程1",) 一定要加上逗号，这样才是一个元祖
p2 = Process(target=task, args=("子进程2",3))
p3 = Process(target=task, args=("子进程3",2))
p_list = [p1,p2,p3]
# 放入列表中去循环
for p in p_list:
p.start()
for p in p_list:
p.join()
# 并发；主程序依次去等待p1、p2、p3
print("主进程",time.time()-start)

 

 Process的其他方法：

p.is_alive()  

from multiprocessing import Process
import os
def task():
print("%s is running"%os.getpid())
if __name__ == "__main__":
p = Process(target=task)
p.start()
print(p.is_alive())
# 判断进程是否还存活；返回bool值

p.join()
print("主进程")
print(p.is_alive())

p.terminate()

from multiprocessing import Process
import os
def task():
print("%s is running"%os.getpid())
if __name__ == "__main__":
p = Process(target=task)
p.start()
p.terminate()
# 终止子进程；p.terminate()也是发给操作系统的一个信号，操作系统什么时候终止子进程你控制不了
print(p.is_alive())
p.join()
print("主进程")
print(p.is_alive())

p.name

from multiprocessing import Process
import os
def task():
print("%s is running"%os.getpid())
if __name__ == "__main__":
p = Process(target=task,name="sub_process")
p.start()
p.join()
print("主进程")
print(p.pid)
print(p.name)
# p.name是进程名，在Process（）实例化的时候name中的名字；如果你不指定系统会选用默认的进程名

 

练习题：

1.

from multiprocessing import Process
n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了
def work():
global n
n=0
print('子进程内: ',n)
if __name__ == '__main__':
p=Process(target=work)
p.start()
# 子进程和主进程的内存空间是互相隔离的；子进程在刚开启的时候，其内存空间的数据确实会复制n=100这个全局变量，然后在work（）中被改成了n==0，但它改的是子进程内存空间中的n,不会影响主进程的n

p.join()
print('主进程内: ',n)
# 运行结果：
# 子进程内:
0
# 主进程内:
100

2. 基于多进程并发的套接字通信

服务端代码：

from socket import *
from multiprocessing import Process
def comm(conn):
#
它的作用是接收、发送数据
# 需要把conn当参数传入函数，因为每次建的链接conn都不一样
while True:
try:
data = conn.recv(1024)
if not data:break
conn.send(data.upper())
except:
break
conn.close()
def server(ip,port):
# 它干的是建链接的活
server = socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind((ip,port))
server.listen(5)
while True:
conn,client_addr = server.accept()
p = Process(target=comm,args=(conn,))
p.start()
server.close()
if __name__ == "__main__":
# windows系统中开进程的操作必须放在 if __name__ == "__main__": 下面
server("127.0.0.1",8080)
"""
这种方式虽然能实现并发，但却是有问题的；因为每有一个客户端发送链接请求，server都需要建一个新的进程，当进程多了以后server容易崩溃
"""

客户端代码：

from socket import *
client = socket(AF_INET,SOCK_STREAM)
client.connect(("127.0.0.1",8080))
# 向IP为 127.0.0.1 Port为8080 的客户端发送建链接请求
while True:
msg = input(">>>").strip()
if not msg:continue
client.send(msg.encode("utf-8"))
data = client.recv(1024)
print(data.decode("utf-8"))
"""
同一个程序运行两次是两个不同的进程，所以这一个client.py 的代码运行多次就是多个进程
"""

守护进程：主进程代码运行结束，子进程随即终止，这样的子进程就是守护进程

关于守护进程有两点需要注意：

　　1. 守护进程会在主进程代码执行结束后立马终止

　　2. 守护进程内无法在开启子进程，否则抛出异常： AssertionError: daemonic processes are not allowed to have children

如果子进程的任务在主进程任务结束后就没有存在的必要了，那么该子进程在开启之前就可以被设置成守护进程。设置方法： p.daemon=True

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__":
p1 = Process(target=foo)
p2 = Process(target=bar)
p1.daemon = True
# 把p1设置成守护进程； # 一点要在p1.start()之前设置为守护进程

p1.start()
p2.start()
print("main-------")
# 只要终端打印出这行内容，那么守护进程p1也就跟着结束掉了

 

互斥锁：

进程之间数据不共享，但是共享同一套文件系统，所以访问同一个文件，或统一打印终端，是没有问题的，而共享带来的是竞争，竞争带来的结果是错乱，如下：

from multiprocessing import Process
import time
def work(name):
print("%s进程 1"%name)
time.sleep(1)
print("%s进程 2"%name)
time.sleep(1)
print("%s进程 3"%name)
if __name__ == "__main__":
for i in range(3):
p = Process(target=work,args=("%s"%i,))
p.start()

 

其中一次的运行结果如下：

由上图可知，终端上的打印结果是错乱的

如何控制，就用到了加锁处理；而互斥锁的意思就是互相排斥，互斥锁的工作原理就是多个任务去争抢同一个资源（例如CPU）： 一个任务抢到锁后，其他任务都要等着，等到这个任务完成后释放锁，其他任务才有可能有一个抢到。 所以互斥锁的原理，就是把并发改成了串行，降低了效率，但保证了数据安全不错乱；如下所示：

from multiprocessing import Process,Lock
import time
def work(name,lock):
lock.acquire()
# 加锁 # 谁抢到锁谁就运行，其他的进程就等着锁释放
print("%s进程 1"%name)
time.sleep(1)
print("%s进程 2"%name)
time.sleep(1)
print("%s进程 3"%name)
lock.release()
# 释放锁
if __name__ == "__main__":
mutex = Lock()
# 父进程先实例化得到一个互斥锁对象
for i in range(3):
p = Process(target=work,args=("%s"%i,mutex))
# 把实例化出来的互斥锁传入各个子进程中；因为你需要子进程争抢你上面实例化出来的这把锁（所有子进程用的都是这同一把锁），而不是子进程刚开启时从父进程的内存空间复制到其内存空间的那把全新的锁（mutex=Lock()这行代码会在子进程中重新运行一次，从而实例化出一个新的锁对象）
p.start()

 

其中一次的运行结果：

模拟抢票：

import json,time
from multiprocessing import Process,Lock
def search(name):
# 查询余票
time.sleep(1)
# 模拟网络延迟
dic = json.load(open("db.json","r",encoding="utf-8"))
print("%s 查到的剩余票数<%s>"%(name,dic["count"]))
def get(name):
# 购票
time.sleep(1)
dic = json.load(open("db.json", "r", encoding="utf-8"))
if dic["count"] > 0:
dic["count"] -= 1
time.sleep(3)
# 模拟付款
json.dump(dic,open("db.json","w",encoding="utf-8"))
print("<%s>购票成功"%name)
def task(name,mutex):
search(name)
# search功能（查票）需要并发完成，即每个人都能够查询到剩余票数； 但购票功能（get）需要加互斥锁串行完成，即同一时间只能有一个人在买票
mutex.acquire()
# 互斥锁要加到get方法（购票）上；通过这种方式，大家的search都是并发执行的，但get功能只有抢到锁的才会执行，其他的人都要等着

get(name)
mutex.release()
if __name__ == "__main__":
mutex = Lock()
for i in range(10):
p = Process(target=task,args=("路人%s"%i,mutex))
p.start()

"""
互斥锁和join的区别：
join是将一个任务整体串行，而互斥锁的好处则是可以将一个任务中的某一段代码串行，比如只让task任务中的get函数串行
"""

一次运行结果：

 

队列：进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。

创建队列的类（底层就是以管道和锁定的方式实现）：

　　Queue([maxsize])：创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递

　　  maxsize是队列中允许最大项数，不写则无大小限制，但需明确：1. 队列内存放的是消息而非大数据；2. 队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

队列主要方法：

q.put()    # 用于插入数据到队列中

q.get()   # 可以从队列读取并且删除元素  #  先进先出原则

from multiprocessing import Queue
q = Queue(3)
# 队列可以放任何类型的数据
q.put("hello")
q.put("world")
q.put(1)
# 把数据放入到q这个“管道、容器”中
print(q.full())
# q.full()判断队列是否满了；返回bool值
# q.put(4) #再放就阻塞住了；锁住
print(q.get())
print(q.get())
print(q.get())
# get：先进先出原则
print(q.empty())
# 判断队列是否空了（读取完了）； 返回bool值
# print(q.get()) #再取就阻塞住；锁住

注意：

# Queue 不能用于 进程池 中进程之间的通信； 进程池 中进程之间的通信 可以使用 Manager 中的 Queue
# 进程间的通信也可以用 Pipe （管道）， Pipe 是简化版的 Queue；Pipe只能适用于 两个 进程之间的通信
# Pipe 的性能高于 Queue
# Manager 也可以用于进程间共享内存的操作，如：Manager().dict() 等

 

生产者、消费者模式：

生产者指的是生产数据的任务，消费者指的是处理数据的任务；生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题；生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产则和消费者的处理能力

from multiprocessing import Process,Queue
import time
def producer(name,queue):
for i in range(3):
res = "%s的包子%s"%(name,i)
time.sleep(0.8)
print("%s 生产了%s" % (name, res))
queue.put(res)
# 把生产的结果放入“容器”中
"""
queue.put(None)不能写在这一步代码里；
因为此处的意思是这个生产者生产完了，但并不意味这别的生产者也生产完了，如果在此处 put（None）,就有可能在容器管道中间夹杂着一个None；
当某个消费者收到了这个None时，它就会break；但其实这个时候管道里面还有数据，这个消费者还不能break
"""
def consumer(name,queue):
while True:
res = queue.get()
# 从“容器”中取“生产”出来的值
if not res:break
# 如果从“容器管道”中收到了None，则break
time.sleep(1)
print("%s消费了%s"%(name,res))
if __name__ == "__main__":
q = Queue() # 在主进程中生成一个“容器”对象
# 生产者们
p1 = Process(target=producer,args=("neo",q))
p2 = Process(target=producer,args=("alex",q))
p3 = Process(target=producer,args=("egon",q))
# 把生产的数据都放入主进程的这个q“容器”中
# 消费者们
c1 = Process(target=consumer,args=("c1",q))
# 取数据都从主进程的这个q“容器”中
c2 = Process(target=consumer,args=("c2",q))
p1.start()
p2.start()
p3.start()
c1.start()
c2.start()
p1.join()
p2.join()
p3.join()
# 这三个join（）确保了生产者都已经生产完毕（并且已经把数据都放到了队列中）

q.put(None)
# 有几个消费者，就用几个 put(None)
q.put(None)
# 确保了每个消费者在数据取完后都能收到None
print("主程序")

 

JoinableQueue的应用：

JoinableQueue（[maxsize]）： 这就像是一个Queue对象，但队列运行项目的使用者（消费者）通知生成者项目已经被成功处理；通知进程是使用共享的信号和条件变量来实现的

方法介绍：

JoinableQueue的示例除了和Queue对象相同的方法之外还具有：

1. q.task_done()  # 使用者使用此方法发送心信号，表示q.get()的返回项已经被处理；如果调用此方法的次数大于从队列中删除项目的数量，将引起ValueError异常；

2. q.join()  # 生产者调用此方法进程阻塞，直到队列中所有的项目都被处理；阻塞将持续到队列中的每个项目均调用q.task_done()方法为止

from multiprocessing import Process,JoinableQueue
import time
def producer(name,queue):
for i in range(3):
res = "%s的包子%s"%(name,i)
time.sleep(0.8)
print("%s 生产了%s" % (name, res))
queue.put(res)
queue.join()
# 等待queue中的数据被取完
# 生产者不结束，queue.join()就不会结束；换句话说，如果生产者结束了，一定是queue.join()结束了；queue.join()结束了，一定是消费者把数据取完了
def consumer(name,queue):
while True:
res = queue.get()
time.sleep(1)
print("%s消费了%s"%(name,res))
queue.task_done()
# 每处理完一个数据，task_done()就会通知queue.join()有一个数据已经被取走了
if __name__ == "__main__":
q = JoinableQueue()
# JoinableQueue的对象能够join（），即能够等待q里面的数据被取完
# 生产者们
p1 = Process(target=producer,args=("neo",q))
p2 = Process(target=producer,args=("alex",q))
p3 = Process(target=producer,args=("egon",q))
# 消费者们
c1 = Process(target=consumer,args=("c1",q))
c2 = Process(target=consumer,args=("c2",q))
c1.daemon = True
c2.daemon = True
# 把c1、c2设置成守护进程；主进程结束后，消费者也就没有存在的意义了

p1.start()
p2.start()
p3.start()
c1.start()
c2.start()
p1.join()
p2.join()
p3.join()
# 这三个join（）确保了生产者都已经执行完毕，也就意味着queue.join()结束了，即消费者把数据取完了
# 此时消费者已经没有存在的意义了，所以消费者应该设置成守护进程
print("主程序")

 

生产者消费者模型总结：

1. 程序中有两类角色：一类负责生产数据（生产者），一类负责处理数据（消费者）

2. 引入生产者消费者模型是为了解决：平衡生产者和消费者之间的速度差，程序解开耦合

3. 生产者消费者模型： 生产者 <----> 队列 <----> 消费者