并发编程——多进程


1、什么是进程?
程序:堆代码
进程:正在运行的程序
进程是一个实体,每个进程都自己的独立内存空间

进程的三个状态:

multiprocessing模块介绍

2、同步和异步:针对的程序运行的状态
同步:提交任务后原地等待任务返回结果,期间不做任何事情
异步:提交任务后,不等待任务返回结果,直接运行下一行代码
3、阻塞和非阻塞:针对程序运行的状态
阻塞:遇到 IO 操作》》》阻塞状态
非阻塞:就绪或者运行状态 >>>>就绪状态

multiprocessing模块介绍
Process([group [, target [, name [, args [, kwargs]]]]]),由该类实例化得到的对象,表示一个子进程中的任务(尚未启动)

强调:
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数,是一个元组形式,必须有逗号
参数介绍:
1 group参数未使用,值始终为None
2
3 target表示调用对象,即子进程要执行的任务
4
5 args表示调用对象的位置参数元组,args=(1,2,'egon',)
6
7 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
8
9 name为子进程的名称
方法介绍:
1 p.start():启动进程,并调用该子进程中的p.run()
 2 p.run():进程启动时运行的方法,正是它去调用target指定的函数,我们自定义类的类中一定要实现该方法
 3
 4 p.terminate():强制终止进程p,不会进行任何清理操作,如果p创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放,进而导致死锁
 5 p.is_alive():如果p仍然运行,返回True
 6
 7 p.join([timeout]):主线程等待p终止(强调:是主线程处于等的状态,而p是处于运行的状态)。timeout是可选的超时时间,需要强调的是,p.join只能join住start开启的进程,而不能join住run开启的进程

4、开启进程的两种方式:
form multiprocessing import Pocess
import time

def task(name)
	print('%s is running '%name)
	time.sleep(3)
	print('%s is over' %name)

if __name__ == '__main__':
	p1=Process(target=task,args=())
	p1.strat()
	print('主程序')
方式1
from multiprocessing import Process
import time


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

    #必须写run方法
    def run(self):
        print('%s is running'%self.name)
        time.sleep(3)
        print('%s is over'%self.name)

if __name__ == '__main__':
    obj = MyProcess('egon')
    obj.start()
    print('主')
方式2

5、进程对象的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__':
    start_time = time.time()
    p_list = []
    for i in range(3):
        p = Process(target=task,args=('子进程%s'%i,i))
        p.start()
        p_list.append(p)
    for i in p_list:
        i.join()

    print('主',time.time()-start_time)
join

6、进程之间空间独立

from multiprocessing import Process

x = 100


def task():
    global x
    x = 1
    print(x)


if __name__ == '__main__':
    p = Process(target=task)
    p.start()
    p.join()
    print('主', x)

'''
F:\python36\python3.exe F:/python_s7/week08/day02/进程间数据是隔离的.py
1
主 100
'''
空间独立

7、进程对象其他相关方法

from multiprocessing import Process,current_process
import time
import os

def task():
    print('%s is running'%os.getpid())
    time.sleep(3)
    print('%s is over'%os.getppid())

if __name__ == '__main__':
    p1 = Process(target=task)
    p1.start()
    # p1.terminate()# 杀死子程序
    time.sleep(3)
    print(p1.is_alive())#判断子程序
    print('主')
其他方法

8、守护进程

主进程创建守护进程

  其一:守护进程会在主进程代码执行结束后就终止

  其二:守护进程内无法再开启子进程,否则抛出异常:AssertionError: daemonic processes are not allowed to have children

from multiprocessing import Process
import time

def task(name):
    print('%s 正活着'%name)
    time.sleep(3)
    print('%s 正常死亡'%name)


if __name__ == '__main__':
    p = Process(target=task,args=('egon总管',))
    p.daemon = True  # 必须在p.start开启进程命令之前声明
    p.start()
    print('皇帝jason正在死亡')

join

9、互斥锁:

牺牲了效率但是保证了数据的安全


锁一定要在主进程中创建,给子进程去用
解决多个进程操作同一份数据,造成数据不安全的情况
加锁会将并发变成串行
锁通常用在对数据操作的部分,并不是对进程全程加锁

mutex.acquire() # 抢锁 一把锁不能同时被多个人使用,没有抢到的人,就一直等待锁释放
buy(i)
mutex.release() # 释放锁

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


# 查票
def search(i):
    with open('info', 'r', encoding='utf-8')as f:
        data = json.load(f)
    print('用户查询余票%s' % data.get('ticket'))


def buy(i):
    # 购买前想查询余票
    with open('info', 'r', encoding='utf-8') as f:
        data = json.load(f)
    # 模拟延迟
    time.sleep(random.randint(1, 3))

    if data.get('ticket') > 0:
        data['ticket'] -= 1
        with open('info', 'w', encoding='utf-8')as f:
            json.dump(data, f)
        print('用户抢票成功%s' % i)
    else:
        print('用户%s抢票失败' % i)


def run(i, mutex):
    search(i)
    mutex.acquire()
    buy(i)
    mutex.release()


if __name__ == '__main__':
    mutex = Lock()
    for i in range(10):
        p = Process(target=run, args=(i, mutex))
        p.start()
互斥锁


10 、进程之间的通讯(队列)
隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的

Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递。
1 q.put方法用以插入数据到队列中,put方法还有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。
2 q.get方法可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常.
3
4 q.get_nowait():同q.get(False)
5 q.put_nowait():同q.put(False)
6
7 q.empty():调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。
8 q.full():调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。
9 q.qsize():返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样

例子:

from multiprocessing import Queue

# 产生一个队列能放5个值
q = Queue(5)

q.put(1)
q.put(2)
q.put(3)
q.put(4)
q.put(5)
print(q.full())

# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.empty())# 判断队列是否为空,需要注意的是,在并发的情况下,这个方法判断不准确!

print(q.get_nowait())# 和get一样取值,但是当队列没有值可以取的时候回报错
print(q.get_nowait())
print(q.get_nowait())
print(q.get_nowait())
print(q.get_nowait())
进程之间通讯

11、生产者与消费者模

"""
生产者消费者模型
    生产者:做包子的  生产数据的
    消费者:买包子的  处理数据的

    解决供需不平衡的问题
        定义一个队列,用来存放固定数量的数据
        解决一个生产者和消费者不需直接打交道,两者都通过队列实现数据传输

    Queue:管道+锁
"""
from multiprocessing import Queue,Process,JoinableQueue
import time
import random


def producer(name,food,q):
    for i in range(5):
        data = '%s生产了%s%s'%(name,food,i)
        time.sleep(random.randint(1,3))
        print(data)
        q.put(data)  # 将生产的数据放入队列中


def consumer(name,q):
    while True:
        data = q.get()
        if data is None:break
        time.sleep(random.randint(1, 3))
        print('%s吃了%s'%(name,data))
        q.task_done()  # 告诉你的队列,你已经将数据取出并且处理完毕


if __name__ == '__main__':
    q = JoinableQueue()  # 生成一个队列对象
    p1 = Process(target=producer,args=('大厨egon','包子',q))
    p2 = Process(target=producer,args=('靓仔tank','生蚝',q))
    c1 = Process(target=consumer,args=('吃货owen',q))
    c2 = Process(target=consumer,args=('坑货kevin',q))
    p1.start()
    p2.start()
    c1.daemon = True
    c2.daemon = True
    c1.start()
    c2.start()
    # 等待生产者生产完所有的数据
    p1.join()
    p2.join()
    # 在生产者生产完数据之后,往队列里面放一个提示性的消息,告诉消费者已经没有数据,你走吧,不要等了
    # q.put(None)
    # q.put(None)
    q.join()  # 等待队列中数据全部取出
    print('主')

  

优质内容筛选与推荐>>
1、黑马程序员-分类(category)
2、Docker使用阿里云docker镜像加速
3、HDU3433 【时间的二分+DP最优解】
4、处理ASProtect的Advanced Import Protect的一些想法和过程(2)
5、杨森翔的书法-23


长按二维码向我转账

受苹果公司新规定影响,微信 iOS 版的赞赏功能被关闭,可通过二维码转账支持公众号。

    阅读
    好看
    已推荐到看一看
    你的朋友可以在“发现”-“看一看”看到你认为好看的文章。
    已取消,“好看”想法已同步删除
    已推荐到看一看 和朋友分享想法
    最多200字,当前共 发送

    已发送

    朋友将在看一看看到

    确定
    分享你的想法...
    取消

    分享想法到看一看

    确定
    最多200字,当前共

    发送中

    网络异常,请稍后重试

    微信扫一扫
    关注该公众号





    联系我们

    欢迎来到TinyMind。

    关于TinyMind的内容或商务合作、网站建议,举报不良信息等均可联系我们。

    TinyMind客服邮箱:support@tinymind.net.cn