进程 multiprocessing Process join Lock Queue

时间 2019-11-24

标签进程 multiprocessing process join lock queue 栏目 SQL 繁體版

原文原文链接

多道技术

1.空间上的复用linux

多个程序公用一套计算机硬件算法

2.时间上的复用 cpu 切换程序+保存程序状态数据库

1.当一个程序遇到IO操做,操做系统会剥夺该程序的cpu执行权限(提升了cpu的利用率,而且不影响程序的执行效率编程

2.当一个程序长时间占用cpu 操做系统也会剥夺该程序的cpu执行权限(下降了程序的执行效率)json

进程

进程是正在运行的程序windows

进程是操做系统中最基本、重要的概念。是多道程序系统出现后，为了刻画系统内部出现的动态状况，描述系统内部各道程序的活动规律引进的一个概念,全部多道程序设计操做系统都创建在进程的基础上。
ps: 同一个程序屡次执行,就会在操做系统中出现两个进程,因此咱们能够同时运行一个软件,分别作不一样的事情也不会混乱

进程调度

时间片转轮法安全

基本思路是让每一个进程在就绪队列中的等待时间与享受服务的时间成比例。在时间片轮转法中，须要将CPU的处理时间分红固定大小的时间片，例如，几十毫秒至几百毫秒。若是一个进程在被调度选中以后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放本身所占有的CPU而排到就绪队列的末尾，等待下一次调度。同时，进程调度程序又去调度当前就绪队列中的第一个进程。

并行与并发

并行: 多个程序同时运行,而且每一个程序有单独的cpu运算(只能在多核计算机上实现)服务器

并发:多个程序看起来是同时运行,其实是由一个cpu来回切换执行的session

**并行**是从微观上，也就是在一个精确的时间片刻，有不一样的程序在执行，这就要求必须有多个处理器。
**并发**是从宏观上，在一个时间段上能够看出是同时执行的，好比一个服务器同时处理多个session。

同步异步 / 阻塞非阻塞

在程序运行的过程当中，因为被操做系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞多线程

同步/异步

表示的是任务的提交方式

同步:任务提交后原地等待任务的执行结果并拿到返回结果后才走,期间不会作任何事(程序层面表现的是卡住了)

异步:任务提交后再也不原地等待,而是继续执行下一行代码(提交的任务结果仍是要的,可是是用骐达方式获取)

import time

def func(name):
    print('%s runing'%name)
    time.sleep(3)

func('waller')  # 同步 任务提交后原地等待函数运行结束后再走下面的代码
print('procedure over')

阻塞/非阻塞

表示的是进程的运行状态

阻塞: 指进程的阻塞态

非阻塞: 指进程的就绪态运行态

multiprocessing模块

multiprocessing模块用来开启子进程，并在子进程中执行咱们定制的任务（好比函数）
multiprocessing模块的功能众多：支持子进程、通讯和共享数据、执行不一样形式的同步，提供了Process、Queue、Pipe、Lock等组件。

Process 建立进程的类

语法

from multiprocessing import Process

代码块

if __name__ == __main___:  # 
    p = Process(target=调用对象, args=调用对象的位置参数元组)  # 建立一个进程对象
    代码块
    p.start()  # 告诉操做系统帮你建立一个进程
    

注意: windows建立进程必定要在 if __name__ == __main___: 代码块内建立,不然会报错



windows建立进程会将代码已模块的方式,从上往下执行一遍,linux会直接将代码完完整整的拷贝一份
建立进程就是在内存中从新开辟了一块内存空间,将运行产生的代码丢进去,一个进程对应砸死内存就是一块独立的内存空间
进程进程之间的数据时隔离的,没法直接交互,可是能够经过某些技术实现间接交互

参数介绍

Process源码:
    
class Process(object):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        ...
# group参数未使用，值始终为None
# target表示调用对象，即子进程要执行的任务
# args表示调用对象的位置参数元组，如:args=(1,2,'egon',)
# kwargs表示调用对象的字典,kwargs={'name':'egon','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开启的进程

建立进程的两种方式

from multiprocessing import Process
import time

def func(name):
    print('%s running'%name)
    time.sleep(3)
    print('over')
print('w')

if __name__ == '__main__':
    # 建立进程对象 
    p = Process(target=func, args=('waller',)) # target=func 是在p进程内存空间中调用执行了 func 函数
    # 告诉操做系统建立一个进程 开辟p进程的内存空间
    p.start()  # 启动p进程
    print('主进程')  # 异步
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self, name):
        super(MyProcess, self).__init__()
        self.name = name
    def run(self):
        print('%s running'%self.name)
        time.sleep(3)
        print('over')

if __name__ == '__main__':
    p = MyProcess('waller')
    p.start()  # 自动运行 run 方法
    print('主进程')

join

主进程等待子进程运行结束后再运行

from multiprocessing import Process
import time

def func(name, i):
    print('%s 进程 running'%name)
    time.sleep(i)
    print('%s 进程 over'%name)

if __name__ == '__main__':
    p1 = Process(target=func, args=('A',1))
    p2 = Process(target=func, args=('B',2))
    p3 = Process(target=func, args=('C',3))
    start_time = time.time()
    p1.start()
    p2.start()
    p3.start()
    p2.join()
    p1.join()
    p3.join()
    print('主进程')
    print(time.time() - start_time)
    '''
    B 进程 running
    A 进程 running
    C 进程 running
    A 进程 over
    B 进程 over
    C 进程 over
    主进程
    3.1728098392486572

    '''

进程之间数据时隔离的

from multiprocessing import Process

x = 100
def func():
    global x
    x = 200
    print(x)

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(x)
    >>> 
    100
    200

进程对象及其余方法

os.getpid() 当前进程的pid
os.getppid() 当前进程的父进程pid
p.terminate() 杀死p进程,告诉操做系统帮你杀死一个进程
p.is_alive() 判断p进程是否存活,返回bool值
def func(name):
    print('%s 进程 running'%name, '%s 进程 %s'%(name, os.getpid()), '父进程%s'%os.getppid())
    time.sleep(3)
    print('%s 进程 over'%name)

if __name__ == '__main__':
    p = Process(target=func, args=('w',))
    p.start()
    print('父进程%s'%os.getpid(), '父父进程%s'%os.getppid())
    >>>
    父进程5224 父父进程10332
    w 进程 running w 进程 1912 父进程5224
    w 进程 over
from multiprocessing import Process
import os
import time

def func(name):
    print('%s 进程 running'%name, '%s 进程 %s'%(name, os.getpid()), '父进程%s'%os.getppid())
    time.sleep(3)
    print('%s 进程 over'%name)

if __name__ == '__main__':
    p = Process(target=func, args=('w',))
    p.start()
    print(p.is_alive())  # True
    p.terminate()  # 告诉操做系统杀死p进程
    time.sleep(1)  # 操做系统的速度没有代码运行的速度快,因此须要睡眠
    print(p.is_alive())  # False
    print('父进程%s'%os.getpid(), '父父进程%s'%os.getppid())

僵尸进程与孤儿进程

父进程回收子进程资源的两种方式

1.join方法

2.父进程进程死亡

全部程序都将步入僵尸进程

孤儿进程: 子进程没死,父进程意外死亡

守护进程

会随着主进程的结束而结束

p.daemon=True
必定要在p.start()前设置,设置p为守护进程,禁止p建立子进程,而且父进程代码执行结束,p即终止运行

注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

from multiprocessing import Process
import time

def func(name):
    print('%s running'%name)
    time.sleep(3)
    print('over')


if __name__ == '__main__':
    p = Process(target=func, args=('waller',))
    p.daemon = True
    p.start()
    time.sleep(1)  # 因为是守护进程,主进程运行结束,子进程就被清空,因此要等待操做系统建立子进程并运行
    print('主进程')

互斥锁

互斥锁
  当多个进程操做同一份数据的时候 会形成数据的错乱
  这个时候必须加锁处理
   将并发变成串行
    虽然下降了效率可是提升了数据的安全
   注意:
    1.锁不要轻易使用 容易形成死锁现象
    2.只在处理数据的部分加锁 不要在全局加锁
  
  锁必须在主进程中产生 交给子进程去使用

from multiprocessing import Process, Lock
import json
import time

# 查票
def search(name):
    with open('data', 'r', encoding='utf-8') as f:
        data = f.read()  # 读出的是json模式的数据
        t = json.loads(data).get('ticket')
        print('用户%s查询余票为:%s'%(name, t))
# search('waller')

# 买票
def buy(name):
    with open('data', 'r', encoding='utf-8') as f:
        data_json = f.read()  # 读出的是json模式的数据
        data = json.loads(data_json)
        t = data.get('ticket')
        # 模拟抢票时间
        time.sleep(2)
        if not t > 0 :
            print('已无票')
            return
        t -= 1
        data['ticket'] = t
        # 跟新数据
        with open('data', 'w', encoding='utf-8') as f:
            json.dump(data, f)  # 将更新后的数据序列化到数据库中
        print('用户%s,购票成功'%name)

# 在进程中调用 search 与 buy 函数
def run(name, mutex):
    search(name)
    mutex.acquire()  # 抢锁
    buy(name)  # 被锁的函数
    mutex.release()  # 释放锁


if __name__ == '__main__':
    # 生成一把锁
    mutex = Lock()
    # 建立5个进程
    for i in range(5):
        p = Process(target=run, args=('waller',mutex))
        p.start()

进程间的通讯

Queue 模块

# 建立队列对象
q = Queue(5)  # 括号内能够传参数,表示是这个队列的最大存储,不传参数默认最大存储
q.put()  # 往队列中添加数据  当队列满了以后,再放数据,不会报错,会原地等待,知道队列中有一个空位置出现
q.full()  # 判断队列是否满了,返回bool值
q.get()  # 从队列中取值  当队列中的值取完后再取,不会报错,程序会阻塞,直到队列中再放入一个值
q.empty()  # 判断队列的值是否取完
q.get_nowait()  # 取值  当队列中没有值可取时,不等待直接报错

from multiprocessing import Queue


# 建立队列对象
q = Queue(5)  # 括号内能够传参数,表示是这个队列的最大存储,不传参数默认最大存储
# 往队列中添加数据
q.put(1)
q.put(2)
# 判断队列是否满了,返回bool值
print(q.full())  # >>> False
q.put(3)
q.put(4)
q.put(5)
print(q.full())  # >>> True
q.put(6)  # 当队列满了以后,再放数据,不会报错,会原地等待,知道队列中有一个空位置出现

# 从队列中取值
print(q.get())
print(q.get())
# 判断队列的值是否取完
print(q.empty())  # >>> False
print(q.get())
print(q.get())
print(q.get())
print(q.get())  # 当队列中的值取完后再取,不会报错,程序会阻塞,直到队列中再放入一个值

print(q.get_nowait())  # 取值 当队列中没有值可取时,不等待直接报错

full()
get_nowait()
empty()
都不适合用于多进程的状况判断,
能够把多进程之间的运行看作是异步,当在判断的那一刻,拿到判断结果的一瞬间,
队列可能会又存入或取出一个值,不许确.

进程间通讯IPC机制

from multiprocessing import Process, Queue

# 状况一: 子进程放数据,父进程取数据
def sub(q):
    q.put('hello')

if __name__ == '__main__':
    # 建立队列对象
    q = Queue()
    p = Process(target=sub, args=(q,))
    p.start()
    print(q.get())  # >>> hello

# def sub(q):
#     print(q.get())  # >>> hello
#
# if __name__ == '__main__':
#     # 建立队列对象
#     q = Queue()
#     p = Process(target=sub, args=(q,))
#     p.start()
#     q.put('hello')


# 状况二: 两个子进程间存取
def sub1(q):
    q.put('hello')

def sub2(q):
    print(q.get())  # >>> hello

if __name__ == '__main__':
    q = Queue()
    s1 = Process(target=sub1, args=(q,))
    s2 = Process(target=sub2, args=(q,))
    s1.start()
    s2.start()

生产者消费者模型

 1.生产者消费者模型

在并发编程中使用生产者和消费者模式可以解决绝大多数并发问题。该模式经过平衡生产线程和消费线程的工做能力来提升程序的总体处理数据的速度。

 2.为何要使用生产者和消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，若是生产者处理速度很快，而消费者处理速度很慢，
那么生产者就必须等待消费者处理完，才能继续生产数据。一样的道理，若是消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题因而引入了生产者和消费者模式。
 3.什么是生产者消费者模式

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

基于队列实现生产者消费者模型

生产者消费者模型总结

    #程序中有两类角色
        一类负责生产数据（生产者）
        一类负责处理数据（消费者）
        
    #引入生产者消费者模型为了解决的问题是：
        平衡生产者与消费者之间的工做能力，从而提升程序总体处理数据的速度
        
    #如何实现：
        生产者<-->队列<——>消费者
    #生产者消费者模型实现类程序的解耦和

产生缘由及解决办法 缘由是：生产者p在生产完后就结束了，可是消费者c在取空了q以后，则一直处于死循环中且卡在q.get()这一步。 解决方式:
　　普通方法:无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就能够break出死循环
　　JoinableQueue 方法: 生产者生产的每一个数据上都作一个标记,消费者每 q.get() 取一个值,都用 q.task_done() 标记一次,q.join()感知队列中的数据所有处理完毕,再最终结束

from multiprocessing import Process, Queue
import time
import random

# 生产者
def producer(name, food, q):
    for i in range(5):
        data = '%s生产%s%s'%(name, food, i+1)
        time.sleep(random.random())
        q.put(data)
        print(data)

# 消费者
def consumer(name, q):
    while True:
        data = q.get()  # 程序停在此处
        if data == None:break
        print('%s吃了%s'%(name, data))
        time.sleep(random.random())


if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=('小明', '包子', q))
    p2 = Process(target=producer, args=('小刘', '馒头', q))
    c1 = Process(target=consumer, args=('哈哈',q))
    c2 = Process(target=consumer, args=('嘻嘻',q))
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    # 此时 生产者已经再也不生产数据,但消费者还在取数据,卡在data = q.get()
    # 解决办法 在生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就能够break出死循环
    p1.join()
    p2.join()  
    q.put(None)
    q.put(None)  # 有两个消费者

代码

调用 JoinableQueue 模块

#JoinableQueue([maxsize])：这就像是一个Queue对象，但队列容许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
#方法介绍：
    JoinableQueue的实例p除了与Queue对象相同的方法以外还具备：
    q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。若是调用此方法的次数大于从队列中删除项目的数量，将引起ValueError异常
    q.join():生产者调用此方法进行阻塞，直到队列中全部的项目均被处理。阻塞将持续到队列中的每一个项目均调用q.task_done（）方法为止

from multiprocessing import Process, JoinableQueue
import time
import random


# 生产者
def producer(name, food, q):
    for i in range(5):
        data = '%s生产%s%s' % (name, food, i + 1)
        time.sleep(random.random())
        q.put(data)
        print(data)


# 消费者
def consumer(name, q):
    while True:
        data = q.get()  # 程序停在此处
        if data == None: break
        print('%s吃了%s' % (name, data))
        time.sleep(random.random())
        q.task_done()


if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer, args=('小明', '包子', q))
    p2 = Process(target=producer, args=('小刘', '馒头', q))
    c1 = Process(target=consumer, args=('哈哈', q))
    c2 = Process(target=consumer, args=('嘻嘻', q))
    p1.start()
    p2.start()
    c1.daemon = True
    c2.daemon = True
    c1.start()
    c2.start()
    # 此时 生产者已经再也不生产数据,但消费者还在取数据,卡在data = q.get()
    # 解决办法 在生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就能够break出死循环
    p1.join()
    p2.join()
    q.join()  # 等到队列中的数据全取出

代码