python并发编程之多进程二

2022-10-16 10:39:17

一，multiprocessing模块介绍

python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。Python提供了multiprocessing。
multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。

　 multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

二，Process类介绍

创建进程的类：

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开启的进程

属性介绍：

1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置

2

3 p.name:进程的名称

4

5 p.pid：进程的pid

6

7 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)

8

9 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

三，Process类的使用

创建并开进程的两种方式

第一种：

from multiprocessing import Process

import random,time

import sys

print(sys.modules)

def eat(name):

    print('%s is eating......'%name)

    time.sleep(random.randint(1,5))

    print('%s had end....'%name)

if __name__ == '__main__':

    p=Process(target=eat,args=('fugui',))#创建进程

    p.start() #开启进程

第二种：

from multiprocessing import Process

import random,time,os

class Eat(Process):

    def __init__(self,name):

        super().__init__()

        self.name=name

    def run(self):

        print(os.getpid(),os.getppid())

        print('%s is eating......' % self.name)

        time.sleep(random.randint(6, 15))

        print('%s had end....' % self.name)

if __name__ == '__main__':

    p=Eat('fugui')

    p1=Eat('xiaojian')

    p2=Eat('liqiang')

    p.start()

    p1.start()

    p2.start()

    print('主进程')

四，守护进程

主进程创建守护进程

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

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

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

from multiprocessing import Process

import random,time

import sys

print(sys.modules)

def eat(name):

    print('%s is eating......'%name)

    time.sleep(random.randint(1,5))

    print('%s had end....'%name)

if __name__ == '__main__':

    p=Process(target=eat,args=('fugui',))#创建进程

    p.daemon()#设置为守护进程

    p.start() #开启进程

五，进程同步（锁）

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,

竞争带来的结果就是错乱，如何控制，就是加锁处理

多个进程共享一个文件

from multiprocessing import Process,Lock

import json,time,random

def search():

    dic = json.load(open('db.txt'))

    print('还剩 %s 张票' % dic['count'])

def get():

    dic = json.load(open('db.txt'))

    if dic['count']>0:

        dic['count']-=1

        time.sleep(random.randint(2,4))

        json.dump(dic,open('db.txt','w'))

        print('购票成功')

def task(mutex):

    search()

    mutex.acquire()#加锁

    get()

    mutex.release()#释放锁

if __name__ == '__main__':

    mutex=Lock()

    for i in range(100):

        p=Process(target=task,args=(mutex,))

        p.start()

模拟抢票加锁

六，队列

用队列模拟生产者消费者模型

from multiprocessing import Process,JoinableQueue

import time,os,random

def producer(q):

    for i in range(1,5):

        res=i

        time.sleep(2)

        q.put(res)

        print('%s 制造了第 %s 个包子'%(os.getpid(),res))

    q.join()

def consumer(q):

    while True:

        res=q.get()

        time.sleep(random.randint(1,3))

        print('%s 吃了了第 %s 个包子' % (os.getpid(), res))

        q.task_done()

if __name__ == '__main__':

    q=JoinableQueue()

    p1=Process(target=producer,args=(q,))

    p2=Process(target=consumer,args=(q,))

    p2.daemon=True

    p1.start()

    p2.start()

    p1.join()

    print('关门')

七，进程池

在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：

很明显需要并发执行的任务通常要远大于核数
一个操作系统不可能无限开启进程，通常有几个核就开几个进程
进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）

例如当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。

我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数...
ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。

from multiprocessing import Pool

import os,time

def task(n):

    print('%s is running'%os.getpid())

    time.sleep(2)

    print('%s is end '%os.getpid())

    return n*n

if __name__ == '__main__':

    p=Pool()

    obj_l=[]

    for i in range(1,7):

        # res=p.apply(task,args=(i,))       #同步运行

        # print('本次任务的结果是 %s'%res)

        res = p.apply_async(task, args=(i,))  #异步运行

        obj_l.append(res)

    p.close()

    p.join()

    print('主')

    for i in obj_l:

        print(i.get())

from multiprocessing import Pool

import requests,os

def get_page(url):

    response=requests.get(url)

    print('%s 正在下载 %s'%(os.getpid(),url))

    return {'url':url,'content':response.text}

def parse_page(res):

    print('%s 正在解析 %s' % (os.getpid(), res['url']))

    with open('db.txt','a')as f:

        dic='url:%s size:%s\n' %(res['url'],len(res['content']))

        f.write(dic)

if __name__ == '__main__':

    urls = [

        'https://www.baidu.com',

        'http://www.openstack.org',

        'https://www.python.org',

        'https://help.github.com/',

        'http://www.sina.com.cn/'

    ]

    p=Pool(4)

    for url in urls:

        obj=p.apply_async(get_page,args=(url,),callback=parse_page)

    p.close()

    p.join()

    print('主')

爬虫案列之回调函数

码农公寓

相关文章