并发编程

Posted 2020-10-23 nerocl

一、进程

1、理论背景

①操作系统背景

• 手工操作—穿孔卡片 
  原理：用户独占计算机，CPU等待手工操作，资源利用和CPU利用效率极低
• 批处理—磁带处理、批处理系统 
  原理：主机与输入机之间增加磁带存储设备，使得计算机可以自动将磁带读入计算机，成批完成作业 
  分类：脱机批处理系统、联机批处理系统
• 多道程序系统 
  原理：指允许多个程序同时进入内存并交替运行 
  特点：宏观上并行、微观上串行
• 分时系统 
  原理：把处理机的运行时间分成很短的时间片，按时间片轮流把处理机分配给各联机作业使用 
  特点：多路性、交互性、独立性、及时性；在进行I/O切换的时候实际上效率相比下降低了
• 实时系统 
  原理：程序独占CPU，为了让程序能够及时响应随机发生的外部事件，并在严格的时间范围内完成对该事件的处理 
  分类：实时控制系统、实时信息处理系统 
  特点：及时响应、高可靠性。
• 通用操作系统 
  定义：具有多种类型操作特征的操作系统。可以同时兼有多道批处理、分时、实时处理的功能，或其中两种以上的功能。
• 操作系统的进一步发展 
  分类：个人计算机操作系统、网络操作系统、分布式操作系统
• 操作系统的作用 
  作用： 
  
  • 1、隐藏了丑陋的硬件调用接口，为应用程序员提供调用硬件资源的更好，更简单，更清晰的模型（系统调用接口）
  • 2、将应用程序对硬件资源的竞态请求变得有序化

②何为进程、何为进程调度

• 进程

• 狭义定义：进程是正在运行的程序的实例
• 广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动
• 进程的意义：是操作系统动态执行的基本单元，操传统操作系统中，进程既是基本的分配单元，也是基本的执行单元
• 操作系统引入进程的原因： 
  理论角度：是对正在运行的程序过程的抽象
实现角度：是一种数据结构，目的是描述动态系统的内在规律，有效管理、调度计算机系统主存储器运行的程序

• 进程的特征：

动态性：进程的实质是程序在多道程序系统中的一次执行过程，进程是动态产生，动态消亡的。 
并发性：任何进程都可以同其他进程一起并发执行 
独立性：进程是一个能独立运行的基本单位，同时也是系统分配资源和调度的独立单位； 
异步性：由于进程间的相互制约，使进程具有执行的间断性，即进程按各自独立的、不可预知的速度向前推进 
结构特征：进程由程序、数据和进程控制块三部分组成。 
多个不同的进程可以包含相同的程序：一个程序在不同的数据集里就构成不同的进程，能得到不同的结果；但是执行过程中，程序不能发生改变。

• • • 进程与程序的区别：

程序是指令和数据的有序集合，其本身没有任何运行的含义，是一个静态的概念。 
而进程是程序在处理机上的一次执行过程，它是一个动态的概念。 
程序可以作为一种软件资料长期存在，而进程是有一定生命期的。 
程序是永久的，进程是暂时的。

• 进程调度

  • 先来先服务调度算法： 
    先来先服务（FCFS）调度算法是一种最简单的调度算法，该算法既可用于作业调度，也可用于进程调度。FCFS算法比较有利于长作业（进程），而不利于短作业（进程）。由此可知，本算法适合于CPU繁忙型作业，而不利于I/O繁忙型的作业（进程）。

  • 短作业优先调度算法： 
    短作业（进程）优先调度算法（SJ/PF）是指对短作业或短进程优先调度的算法，该算法既可用于作业调度，也可用于进程调度。但其对长作业不利；不能保证紧迫性作业（进程）被及时处理；作业的长短只是被估算出来的。

  • 时间片轮转法： 
    时间片轮转(Round Robin，RR)法的基本思路是让每个进程在就绪队列中的等待时间与享受服务的时间成比例。在时间片轮转法中，需要将CPU的处理时间分成固定大小的时间片，如果一个进程在被调度选中之后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放自己所占有的CPU而排到就绪队列的末尾，等待下一次调度。同时，进程调度程序又去调度当前就绪队列中的第一个进程。

  • 多级反馈队列： 
    应设置多个就绪队列，并为各个队列赋予不同的优先级。当一个新进程进入内存后，首先将它放入第一队列的末尾，按FCFS原则排队等待调度。当轮到该进程执行时，如它能在该时间片内完成，便可准备撤离系统；如果它在一个时间片结束时尚未完成，调度程序便将该进程转入第二队列的末尾按照FCFS原则等待调度执行，以此类推，而且仅当第一队列空闲时，调度程序才调度第二队列中的进程运行

③进程的并发、并行

• 并行：是指两者同时执行
• 并发：是指资源有限的情况下，两者交替轮流使用资源，以便提高效率
• 区别：并行是从微观上，也就是在一个精确的时间片刻，有不同的程序在执行，这就要求必须有多个处理器；并发是从宏观上，在一个时间段上可以看出是同时执行的，比如一个服务器同时处理多个session。 
  

④同步、异步、阻塞、非阻塞

• 同步： 
  系统功能调用时，在没有得到结果之前，该调用就不会返回。按照这个定义，其实绝大多数函数都是同步调用。但是一般而言，我们在说同步、异步的时候，特指那些需要其他部件协作或者需要一定时间完成的任务。

• 异步： 
  异步的概念和同步相对。当一个异步功能调用发出后，调用者不能立刻得到结果。当该异步功能完成后，通过状态、通知或回调来通知调用者。如果异步功能用状态来通知，那么调用者就需要每隔一定时间检查一次，效率就很低（有些初学多线程编程的人，总喜欢用一个循环去检查某个变量的值，这其实是一 种很严重的错误）。如果是使用通知的方式，效率则很高，因为异步功能几乎不需要做额外的操作。至于回调函数，其实和通知没太多区别。

• 阻塞： 
  阻塞调用是指调用结果返回之前，当前线程会被挂起（如遇到io操作）。函数只有在得到结果之后才会将阻塞的线程激活。有人也许会把阻塞调用和同步调用等同起来，实际上他是不同的。对于同步调用来说，很多时候当前线程还是激活的，只是从逻辑上当前函数没有返回而已。

• 非阻塞： 
  非阻塞和阻塞的概念相对应，指在不能立刻得到结果之前也会立刻返回，同时该函数不会阻塞当前线程。

• 总结：

  • 同步与异步针对的是函数/任务的调用方式：同步就是当一个进程发起一个函数（任务）调用的时候，一直等到函数（任务）完成，而进程继续处于激活状态。而异步情况下是当一个进程发起一个函数（任务）调用的时候，不会等函数返回，而是继续往下执行当，函数返回的时候通过状态、通知、事件等方式通知进程任务完成。
  • 阻塞与非阻塞针对的是进程或线程：阻塞是当请求不能满足的时候就将进程挂起，而非阻塞则不会阻塞当前进程

• 同步阻塞形式： 
  效率最低。一直等待进程结束，其他什么也不做

• 异步阻塞形式： 
  异步操作是可以被阻塞住的，只不过它不是在处理消息时阻塞，而是在等待消息通知时被阻塞

• 同步非阻塞形式： 
  实际上是效率低下的，因为程序需要在这两种不同的行为之间来回的切换，效率浪费在了程序切换中

• 异步非阻塞形式： 
  效率较高，程序没有在两种不同的操作中来回切换

• 总结： 
  同步和阻塞混淆很容易混淆，因为很多时候同步操作会以阻塞的形式表现出来，同样很多人也会把异步和非阻塞混淆，因为异步操作一般都不会在真正的I/O操作处被阻塞。

⑤进程的创建、结束(了解)

• 进程的创建的4种形式：

  • 系统初始化（查看进程linux中用ps命令，windows中用任务管理器，前台进程负责与用户交互，后台运行的进程与用户无关，运行在后台并且只在需要时才唤醒的进程，称为守护进程，如电子邮件、web页面、新闻、打印）
  • 一个进程在运行过程中开启了子进程（如nginx开启多进程，os.fork,subprocess.Popen等）
  • 用户的交互式请求，而创建一个新进程（用户双击操作）
  • 一个批处理作业的初始化（只在大型机的批处理系统中应用）

• 进程的结束

  • 正常退出（自愿，如用户点击交互式页面的叉号，或程序执行完毕调用发起系统调用正常退出，在linux中用exit，在windows中用ExitProcess）
  • 出错退出（自愿，python a.py中a.py不存在）
  • 严重错误（非自愿，执行非法指令，如引用不存在的内存，I/O等，可以捕捉异常，try…except…）
  • 被其他进程杀死（非自愿，如kill -9 PID）

2、python中的进程应用——multiprocessing模块

①multiprocessing模块简介：

multiprocess不是一个模块而是python中一个操作、管理进程的包，这个包中几乎包含了和进程有关的所有子模块。以下将分为：进程创建部分、进程同步部分、进程间通信部分、进程池部分和进程之间数据共享来介绍

②进程创建部分

• process模块语法简介 
  Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）

  • 强调：

需要使用关键字的方式来指定参数
args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

• • • 参数介绍：

group参数未使用，值始终为None
target表示调用对象，即子进程要执行的任务
args表示调用对象的位置参数元组，args=(1,2,’yang’,)
kwargs表示调用对象的字典,kwargs={‘name’:’yang’,’age’:18}
name为子进程的名称

• • • 子进程常用方法

p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

• ips
  在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在import的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来，import 的时候，就不会递归运行了。

• 面向函数——创建子进程实现并发

import os
from multiprocessing import Process

def func():
    print(‘this is subprocess:{}‘.format(os.getpid()))

if __name__ == ‘__main__‘:
    p = Process(target=func,args=())
    p.start()
    print(‘this is parent_process:{}‘.format(os.getpid()))

• 面向函数——创建多个子进程

import os
from multiprocessing import Process

def func(i):
    print(‘{}、this is subprocess:{},father:{}‘.format(i,os.getpid(), os.getppid()))

if __name__ == ‘__main__‘:
    for i in range(5):
        p = Process(target=func, args=(i+1,))
        p.start()
    print(‘this is parent_process:{}‘.format(os.getpid()))

• 面向函数——多个进程同时运行

import os
import time
from multiprocessing import Process

def func(i):
    time.sleep(0.1)
    print(‘{}、this is subprocess:{},father:{}‘.format(i,os.getpid(), os.getppid()))

if __name__ == ‘__main__‘:
    p_lst = []
    for i in range(5):
        p = Process(target=func, args=(i+1,))
        p.start()
        p_lst.append(p)
    [j.join() for j in p_lst]
    print(‘this is parent_process:{}‘.format(os.getpid()))

• 面向对象——创建进程

import os
from multiprocessing import Process

class MyProcess(Process):
    def __init__(self, something):
        super().__init__()
        self.something = something

    def run(self):
        print(‘this is subprocess：{}‘.format(os.getpid()), self.something)

if __name__ == ‘__main__‘:
    p1 = MyProcess(‘我是参数‘)
    p2 = MyProcess(‘我是参数‘)
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print(‘this is parent_process:{}‘.format(os.getpid()))

• 进程间数据隔离问题

from multiprocessing import Process

def work():
    global n
    n=0
    print(‘子进程内: ‘,n)

if __name__ == ‘__main__‘:
    n = 100
    p=Process(target=work)
    p.start()
    print(‘主进程内: ‘,n)

• 守护进程：会随着主进程的结束而结束 
  主进程创建守护进程 
  其一：守护进程会在主进程代码执行结束后就终止 
  其二：守护进程内无法再开启子进程,否则抛出AssertionError异常 
  注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

③进程同步部分(multiprocess.Lock、multiprocess.Semaphore、multiprocess.Event)

• 锁multiprocess.Lock

  • 加锁可以保证多个进程修改同一块数据时，同时刻只能有一个任务可以进行修改，即串行的修改，这样做虽然牺牲了速度却保证了数据安全。所以可以mutiprocessing模块为我们提供更好的的基于消息的IPC通信机制：队列和管道

  • 应用实例

def search(i):
    with open(‘ticket‘, mode=‘r‘, encoding=‘utf-8‘) as f:
        print(i, ‘剩余票数：{}‘.format(json.load(f)[‘ticket_num‘]))


def get(i):
    with open(‘ticket‘, mode=‘r‘, encoding=‘utf-8‘) as f:
        num = json.load(f)[‘ticket_num‘]
    time.sleep(random.random())
    if num > 0:
        with open(‘ticket‘, mode=‘w‘, encoding=‘utf-8‘) as f_w:
            json.dump({‘ticket_num‘: num - 1}, f_w)
            print(‘{}:买到票了！‘.format(i))
    else:
        print(‘{}:没有买到票！‘.format(i))

def task(i, lock):
    search(i)
    lock.acquire()
    get(i)
    lock.release()

if __name__ == ‘__main__‘:
    with open(‘ticket‘,mode=‘w‘) as f_1:
        json.dump({‘ticket_num‘: 5}, f_1)
    lock = Lock()
    for i in range(10):
        p = Process(target=task, args=(i, lock,))
        p.start()

信号量multiprocess.Semaphore

• 介绍：互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据

• 注意：信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1；当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念。

• 应用实例

def sing(i,sem):
    sem.acquire()
    print(‘{}:进入>>>>‘.format(i))
    time.sleep(random.randint(1,3))
    print(‘{}:<<<<出‘.format(i))
    sem.release()

if __name__ == ‘__main__‘:
    sem = Semaphore(4)
    for i in range(10):
        p = Process(target=sing,args=(i,sem))
        p.start()

事件multiprocess.Event

• 介绍：

事件用于主线程控制其他线程的执行，事件主要提供了三个方法 e.set()、e.wait()、e.clear()、e.is_set()事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
clear：将“Flag”设置为False
set：将“Flag”设置为True

• 应用实例

def traffic_light(e):
    while True:
        if e.is_set():
            time.sleep(1)
            print(‘红灯亮‘)
            e.clear()
        else:
            time.sleep(1)
            print(‘绿灯亮‘)
            e.set()

def car(i, e):
    e.wait()
    print(‘{}:车通过...‘.format(i))

if __name__ == ‘__main__‘:
    e = Event()
    traffic = Process(target=traffic_light, args=(e,))
    traffic.daemon = True
    traffic.start()
    for i in range(30):
        if i % 6:
            time.sleep(random.random())
        car_obj = Process(target=car, args=(i, e))
        car_obj.start()

④进程间通信部分(multiprocess.Queue、multiprocess.Pipe)

• 进程间通信：IPC(Inter-Process Communication)

• 队列multiprocess.Queue

  • 介绍：创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

  • 语法：

Queue([maxsize]) ：创建共享的进程队列

参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。（底层队列使用管道和锁定实现）

• Queue的实例q具有以下方法：

q.get( [ block [ ,timeout ] ] )

返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。
q.get_nowait( )

同q.get(False)方法。
q.put(item [, block [,timeout ] ] )

将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常(定义在Queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
q.qsize()

返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。
q.empty()

如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。
q.full()

如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty()方法）

• 管道multiprocess.Pipe

⑤进程池部分

⑥进程之间数据共享