进程

1. 进程

引入模块

from multiprocessing import Process    #谋定噗rua赛斯听            #噗rua赛斯

进程之间是空间隔离的,不共享资源

进程的两种创建方法

创建进程的第一种方式:

    p1 = Process(target=func1, args=(1,))　　　　　　#target(他给特)#args(啊渴死)
    p1.start()　　　　　　　　　　　　　　　　　　　　　　　　#start四大特

创建进行的第二种方式:

　　自己定义一个类,继承Process类,必须写一个run方法,想传参数,自行写init方法,然后执行super父类的init方法

class MyProcess(Process):
    def __init__(self,n,name):
        super().__init__()
        self.n = n
        self.name = name
    def run(self):
        # print(1+1)
        # print(123)
        print('子进程的进程ID',os.getpid())
        print('你看看n>>',self.n)

if __name__ == '__main__':
    p1 = MyProcess(100,name='子进程1')
    p1.start() #给操作系统发送创建进程的指令,子进程创建好之后,要被执行,执行的时候就会执行run方法
    print('p1.name',p1.name)
    print('p1.pid',p1.pid)
    print('主进程结束')

　进程的其他方法　

import time
from multiprocessing import Process
def func1():
    time.sleep(2)
    print()
    print('子进程')
if __name__ == '__main__':
    p1 = Process(target=func1,)
    p1.start()
    p1.terminate() # 给操作系统发了一个关闭p1子进程的信号,关闭进程　　　　　　　　terminate(涛妹雷特)
    time.sleep(1)　　　　　　　　　　　　　　四离谱
    print('进程是否还活着:',p1.is_alive())#判断进程是否存活　　　　　　　　　　is_alive(A子蓝付)
    print(p1.pid)
    print('主进程结束')

class MyProcess(Process):
    def __init__(self,n,name):
        super().__init__()
        self.n = n
        self.name = name
    def run(self):
        # print(1+1)
        # print(123)
        print('子进程的进程ID',os.getpid())
        print('你看看n>>',self.n)
if __name__ == '__main__':
    p1 = MyProcess(100,name='子进程1')
    p1.start() #给操作系统发送创建进程的指令,子进程创建好之后,要被执行,执行的时候就会执行run方法
    print('p1.name',p1.name)
    print('p1.pid',p1.pid)
    print('主进程结束')

join方法

global_num = 100
def func1():
    time.sleep(2)
    global global_num
    global_num = 0
    print('子进程全局变量>>>',global_num)
if __name__ == '__main__':
    p1 = Process(target=func1,)
    p1.start()
    print('子进程执行')
    #time.sleep(3)
    p1.join()  #阻塞住,等待你的p1子进程执行结束,主进程的程序才能从这里继续往下执行
    print('主进程的全局变量>>>',global_num)

global_num = 100
def func1():
    start_time = time.time()
    time.sleep(2)
    global global_num
    global_num = 0
    print('子进程全局变量>>>',global_num)
    end_time = time.time()
    print(end_time - start_time)
if __name__ == '__main__':
    p1 = Process(target=func1,)
    p1.start()
    print('子进程执行')
    #time.sleep(3)
    p1.join()  #阻塞住,等待你的p1子进程执行结束,主进程的程序才能从这里继续往下执行
    print('主进程的全局变量>>>',global_num)

#for循环在创建进程中的应用
def fun1(n):
    time.sleep(1)
    print(n)

if __name__ == '__main__':
    pro_list = []
    for i in range(10):
        p1 = Process(target=fun1,args=(i,))
        p1.start()
        pro_list.append(p1)

    for p in pro_list:
        p.join()
    print('主进程结束')

守护进程   一定要在start之前设置守护进程

import time
import os
from multiprocessing import Process
def func1():
    time.sleep(5)
    print(os.getpid())
    print('子进程')

if __name__ == '__main__':
    p1 = Process(target=func1,)
    p1.daemon = True#将p1子进程设置为守护进程　　　　　　　　daemon(地们)
    p1.start()
    # print('主进程的ID',os.getpid())
    print('主进程结束')

三、进程同步(锁)

问题1：为什么要加进程锁？

       线程锁是为了在线程不安全的时候，为一段代码加上锁来控制实现线程安全，即线程间数据隔离；

       进程间的数据本来就是隔离的，所以一般不用加锁，当进程间共用某个数据的时候需要加锁；

Look(唠嗑)模块　　　　acquire(额快也)#加锁　　　　release(蕊李四)还锁

加锁的另种形式:     with

import json
import time
import random
from multiprocessing import Process,Lock

def get_ticket(i,ticket_lock):
    print('我们都到齐了,大家预备!!123')
    #所有进程异步执行,到这里等待,同时再去抢下面的代码执行
    time.sleep(1)
    ticket_lock.acquire()  #这里有个门,只有一个人能够抢到这个钥匙,加锁
    with open('ticket','r') as f:
        #将文件数据load为字典类型的数据
        last_ticket_info = json.load(f)
    #查看一下余票信息
    last_ticket = last_ticket_info['count']
    #如果看到余票大于0,说明你可以抢到票
    if last_ticket > 0:
        #模拟网络延迟时间
        time.sleep(random.random())
        #抢到一张票就减去1
        last_ticket = last_ticket - 1
        last_ticket_info['count'] = last_ticket
        #将修改后的票数写回文件
        with open('ticket','w') as f:
            #通过json.dump方法来写回文件,字符串的形式
            json.dump(last_ticket_info,f)
        print('%s号抢到了,丫nb!' % i)
    else:
        print('%s号傻逼,没票啦,明年再来!' % i)
    #释放锁,也就是还钥匙的操作
    ticket_lock.release()
if __name__ == '__main__':
    #创建一个锁
    ticket_lock = Lock()
    for i in range(10):
        #将锁作为参数传给每个进程,因为每个进程都需要通过锁来进行限制,同步
        p = Process(target=get_ticket,args=(i,ticket_lock,))
        p.start()

四、信号量(了解)　　　　

Semaphore(赛呢fao)

import time
import random
from multiprocessing import Process,Semaphore

def dbj(i,s):
    s.acquire()
    print('%s号男主人公来洗脚'%i)
    print('-------------')
    time.sleep(random.randrange(3,6))
    # print(time.time())
    s.release()

if __name__ == '__main__':
    s = Semaphore(4) #创建一个计数器,每次acquire就减1,直到减到0,那么上面的任务只有4个在同时异步的执行,后面的进程需要等待.
    for i in range(10):
        p1 = Process(target=dbj,args=(i,s,))
        p1.start()

五、事件(了解)

Event(额外特)

e.clear(可累儿)将e改成False

e.wait(威特)等待

e.set(赛特)将e是True

e.is_set()查看当前e的状态是True或者False

import time
import random
from multiprocessing import Process,Event

#模拟红绿灯执行状态的函数
def traffic_lights(e):
    while 1:
        print('红灯啦')
        time.sleep(5)
        e.set()
        print('绿灯亮')
        time.sleep(3)
        e.clear()  #将e改为了False
def car(i,e):

    if not e.is_set(): #新来的车看到是红灯
        print('我们在等待.....')
        e.wait()
        print('走你')
    else:
        print('可以走了!!!')

if __name__ == '__main__':
    e = Event()
    hld = Process(target=traffic_lights,args=(e,))
    hld.start()
    while 1:
        time.sleep(0.5)
        #创建10个车
        for i in range(3):
            # time.sleep(random.randrange(1,3))
            p1 = Process(target=car,args=(i,e,))
            p1.start()

复制代码
from multiprocessing import Process,Semaphore,Event
import time,random

e = Event() #创建一个事件对象
print(e.is_set())  #is_set()查看一个事件的状态，默认为False，可通过set方法改为True
print('look here！')
# e.set()          #将is_set()的状态改为True。
# print(e.is_set())#is_set()查看一个事件的状态，默认为False，可通过set方法改为Tr
# e.clear()        #将is_set()的状态改为False
# print(e.is_set())#is_set()查看一个事件的状态，默认为False，可通过set方法改为Tr
e.wait()           #根据is_set()的状态结果来决定是否在这阻塞住，is_set()=False那么就阻塞，is_set()=True就不阻塞
print('give me！！')

#set和clear  修改事件的状态 set-->True   clear-->False
#is_set     用来查看一个事件的状态
#wait       依据事件的状态来决定是否阻塞 False-->阻塞  True-->不阻塞

六、进程间通信 IPC(重要)

　　队列:(重点)

full(fao)

from multiprocessing import Process,Queue
#先进后出
'''
要注意的事项
q.full()队列满了返回True,不满返回False
队列为空的时候,get会阻塞          盖特
put超出了队列长度,你put插入数据的时候会阻塞     破特
print('>>>',q.empty())  #不可信,队列空了返回True,不为空返回False
'''
q = Queue(3)
q.put(1)#往队列里写入3个数据
q.put(2)
q.put(3)
print(q.get())#获取值
#可以用异常踹一下
while 1:
    try:
        q.get(False)  #queue.Empty
    except:
        print('队列目前是空的')

队列实现进程间的通信

import time
from multiprocessing import Process,Queue

def girl(q):
    print('来自boy的消息',q.get())
    print('来自校领导的凝视',q.get())
def boy(q):
    q.put('约吗')

if __name__ == '__main__':
    q = Queue(5)
    boy_p = Process(target=boy,args=(q,))
    girl_p = Process(target=girl,args=(q,))
    boy_p.start()
    girl_p.start()
    time.sleep(1)
    q.put('好好工作,别乱搞')

生产者消费者模型

import time
from multiprocessing import Process,Queue

def producer(q):
    for i in range(1,11):
        time.sleep(1)
        print('生产了包子%s号' % i)
        q.put(i)
    q.put(None)  #针对第三个版本的消费者,往队列里面加了一个结束信号
def consumer(q):
    while 1:
        time.sleep(2)
        s = q.get()
        if s == None:
            break
        else:
            print('消费者吃了%s包子' % s)

if __name__ == '__main__':
    #通过队列来模拟缓冲区,大小设置为20
    q = Queue(20)
    #生产者进程
    pro_p = Process(target=producer,args=(q,))
    pro_p.start()
    #消费者进程
    con_p = Process(target=consumer,args=(q,))
    con_p.start()

　　

生产者消费者模型主进程发送结束信号

import time
from multiprocessing import Process,Queue

def producer(q):
    for i in range(1,11):
        time.sleep(1)
        print('生产了包子%s号' % i)
        q.put(i)

def consumer(q):
    while 1:
        time.sleep(2)
        s = q.get()
        if s == None:
            break
        else:
            print('消费者吃了%s包子' % s)

if __name__ == '__main__':
    #通过队列来模拟缓冲区,大小设置为20
    q = Queue(20)
    #生产者进程
    pro_p = Process(target=producer,args=(q,))
    pro_p.start()
    #消费者进程
    con_p = Process(target=consumer,args=(q,))
    con_p.start()
    pro_p.join()

    q.put(None)　　

JoinableQueue的生产着消费者模型     Joinable(准呢包)Queue   张波Q

 q.task_done(她可死但)     给q对象发送一个任务结束的信号

import time
from multiprocessing import Process,Queue,JoinableQueue

def producer(q):
    for i in range(1,11):
        time.sleep(0.5)
        print('生产了包子%s号' % i)
        q.put(i)
    q.join()
    print('在这里等你')
def consumer(q):
    while 1:
        time.sleep(1)
        s = q.get()
        print('消费者吃了%s包子' % s)
        q.task_done()  #给q对象发送一个任务结束的信号

if __name__ == '__main__':
    #通过队列来模拟缓冲区,大小设置为20
    q = JoinableQueue(20)
    #生产者进程
    pro_p = Process(target=producer,args=(q,))
    pro_p.start()
    #消费者进程
    con_p = Process(target=consumer,args=(q,))
    con_p.daemon = True #
    con_p.start()
    pro_p.join()
    print('主进程结束')　　

管道:

　　进程间通信(IPC)方式二:管道(了解即可)

 Pipe(牌破)

from multiprocessing import Process,Pipe
def func1(conn1,conn2):
    try:
        msg = conn2.recv()
        print('>>>',msg)
        #如果管道一端关闭了,那么另外一端在接收消息的时候会报错
        msg2 = conn2.recv() #EOFError
    except EOFError:
        print('对方管道一端已经关闭')
        conn2.close()

if __name__ == '__main__':
    conn1, conn2 = Pipe()
    p = Process(target=func1,args=(conn1,conn2,))
    p.start()
    conn1.send('小鬼!')
    conn1.close()
    # conn1.recv()  #OSError: handle is closed

数据共享:(了解)

Lock(啦个)

from multiprocessing import Process,Manager,Lock

def func(m_dic,ml):
    #不加锁的情况会出现数据错乱
    # m_dic['count'] -= 1
    #加锁,这是另外一种加锁形式
    with ml:
        m_dic['count'] -= 1

    #等同
    # ml.acquire()
    # m_dic['count'] -= 1
    # ml.release()

if __name__ == '__main__':
    m = Manager()
    ml = Lock()
    m_dic = m.dict({'count':100})
    # print('主进程', m_dic)
    p_list = []
    #开启20个进程来对共享数据进行修改
    for i in range(20):
        p1 = Process(target=func,args=(m_dic,ml,))
        p1.start()
        p_list.append(p1)
    [ppp.join() for ppp in p_list]

    print('主进程',m_dic)

进程池和mutiprocess.Poll(噗奥)

　　

为什么要有进程池?进程池的概念。

　　在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程（空间，变量，文件信息等等的内容）也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，维护一个很大的进程列表的同时，调度的时候，还需要进行切换并且记录每个进程的执行节点，也就是记录上下文（各种变量等等乱七八糟的东西，虽然你看不到，但是操作系统都要做），这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。就看我们上面的一些代码例子，你会发现有些程序是不是执行的时候比较慢才出结果，就是这个原因，那么我们要怎么做呢？

　　在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果

p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()'''

p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。'''
    
p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成

P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

 apply(额扑来)      apply_async(额扑来森科)   Map(麦坡)  

import time
from multiprocessing import Pool,Process

#针对range(100)这种参数的
# def func(n):
#     for i in range(3):
#         print(n + 1)

def func(n):
    print(n)
    # 结果：
    #     (1, 2)
    #     alex
def func2(n):
    for i in range(3):
        print(n - 1)
if __name__ == '__main__':
    #1.进程池的模式
    s1 = time.time()  #我们计算一下开多进程和进程池的执行效率
    poll = Pool(5) #创建含有5个进程的进程池
    # poll.map(func,range(100)) #异步调用进程，开启100个任务,map自带join的功能
    poll.map(func,[(1,2),'alex']) #异步调用进程，开启100个任务,map自带join的功能
    # poll.map(func2,range(100))  #如果想让进程池完成不同的任务，可以直接这样搞
    #map只限于接收一个可迭代的数据类型参数，列表啊，元祖啊等等，如果想做其他的参数之类的操作，需要用后面我们要学的方法。
    # t1 = time.time() - s1
    #
    # #2.多进程的模式
    # s2 = time.time()
    # p_list = []
    # for i in range(100):
    #     p = Process(target=func,args=(i,))
    #     p_list.append(p)
    #     p.start()
    # [pp.join() for pp in p_list]
    # t2 = time.time() - s2
    #
    # print('t1>>',t1) #结果：0.5146853923797607s 进程池的效率高
    # print('t2>>',t2) #结果：12.092015027999878s

同步与异步两种执行方式:

同步:

import time
from multiprocessing import Process,Pool


def fun(i):
    time.sleep(0.5)
    # print(i)
    return i**2
if __name__ == '__main__':
    p = Pool(4)
    for i in range(10):
        res = p.apply(fun,args=(i,))  #同步执行的方法,他会等待你的任务的返回结果,
        print(res)

　　

异步:

import time
from multiprocessing import Process,Pool

def fun(i):
    time.sleep(1)
    print(i)
    return i**2
if __name__ == '__main__':
    p = Pool(4)
    res_list = []
    for i in range(10):
        res = p.apply_async(fun,args=(i,))  
        res_list.append(res)

    p.close()  # 不是关闭进程池,而是不允许再有其他任务来使用进程池
    p.join()   # 这是感知进程池中任务的方法,进程池中所有的进程随着主进程的结束而结束了,等待进程池的任务全部执行完
    #循环打印结果
    for e_res in res_list:
        print('结果:',e_res.get())

    print('主进程结束')

　　

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数，
这是进程池特有的，普通进程没有这个机制，但是我们也可以通过进程通信来拿到返回值，进程池的这个回调也是进程通信的机制完成的。
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果

import os
from multiprocessing import Pool

def func1(n):
    print('func1>>',os.getpid())
    # print('func1')
    return n*n


def func2(nn):
    print('func2>>',os.getpid())
    # print('func2')
    print(nn)
    # import time
    # time.sleep(0.5)
if __name__ == '__main__':
    print('主进程：',os.getpid())
    p = Pool(4)
    p.apply_async(func1,args=(10,),callback=func2)
    p.close()
    p.join()

　　

回调函数在写的时候注意一点，回调函数的形参执行有一个，如果你的执行函数有多个返回值，那么也可以被回调函数的这一个形参接收，接收的是一个元祖，包含着你执行函数的所有返回值。