python --- mulitprocessing(多进程)模块使用

简介: 1. 什么是进程?  进程(Process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和调度的基本单位,是操作系统结构的基础。在早期面向进程设计的计算机结构中,进程是程序的基本执行实体;在当代面向线程设计的计算机结构中,进程是线程的容器。

1. 什么是进程?

  进程(Process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和调度的基本单位,是操作系统结构的基础。在早期面向进程设计的计算机结构中,进程是程序的基本执行实体;在当代面向线程设计的计算机结构中,进程是线程的容器。

一个进程至少包含一个线程。

2. 在python中有了多线程编程为何还需要多进程编程?

  在python中由于有GIL(全局解释器锁)的存在,在任一时刻只有一个线程在运行(无论你的CPU是多少核),无法实现真正的多线程。那么该如何让python程序真正的并行运行呢?答案就是不要使用多线程,使用多进程。python标准库提供了multiprocessing模块(multiprocessing是一个和threading模块类似,提供API,生成进程的模块。multiprocessing包提供本地和远程并发,通过使用子进程而不是线程有效地转移全局解释器锁。),它的API几乎复制了threading模块的API,当然它还有一行threading模块没有的API。

 

例一(multiprocessing模块的简单使用):

 1 import multiprocessing,time
 2 
 3 class Task(multiprocessing.Process):
 4     def __init__(self):
 5         super(Task, self).__init__()
 6 
 7     def run(self):
 8         print("Process---%s" % self.name)
 9         time.sleep(2)
10 
11 
12 if __name__ == "__main__":
13     for i in range(1, 8+1):
14         t = Task()
15         t.start()
View Code

  注:由于multiprocessing模块基本的API同threading模块,就不挨个演示了,本文主要讲解multiprocessing模块不同于threading模块的API的使用。要了解其他同threading模块相同的API的使用,可参见:http://www.cnblogs.com/God-Li/p/7732407.html

 

multiprocessing.Process源码:

class Process(object):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        self.name = ''
        self.daemon = False      #守护进程标志,必须在start()之前设置
        self.authkey = None      #The process’s authentication key (a byte string).
        self.exitcode = None     #The child’s exit code. This will be None if the process has not yet terminated. A negative value -N indicates that the child was terminated by signal N.
        self.ident = 0
        self.pid = 0          #进程ID。在生成进程之前,这将是Non。
        self.sentinel = None   #A numeric handle of a system object which will become “ready” when the process ends.

    def run(self):
        pass

    def start(self):
        pass

    def terminate(self):
        """
        Terminate the process. On Unix this is done using the SIGTERM signal; on Windows TerminateProcess() is used. 
        Note that exit handlers and finally clauses, etc., will not be executed.
        Note that descendant processes of the process will not be terminated – they will simply become orphaned.
        :return: 
        """
        pass

    def join(self, timeout=None):
        pass

    def is_alive(self):
        return False

 

multiprocessing模块中的队列:

  class multiprocessing.Queue([maxsize])实现除task_done()join()之外的queue.Queue的所有方法,下面列出queue.Queue中没有的方法:

class multiprocessing.Queue([maxsize])
    close()
    """
    指示当前进程不会在此队列上放置更多数据。
    The background thread will quit once it has flushed all buffered data to the pipe.
    当队列被垃圾回收时,这被自动调用。
    """

    join_thread()
    """
    加入后台线程。这只能在调用close()之后使用。它阻塞直到后台线程退出,确保缓冲区中的所有数据都已刷新到pipe。
    默认情况下,如果进程不是队列的创建者,那么在退出时它将尝试加入队列的后台线程。
    该进程可以调用cancel_join_thread()使join_thread()不执行任何操作
    """

    cancel_join_thread()
    """
    使join_thread()不执行任何操作
    """

  class multiprocessing.SimpleQueue是class multiprocessing.Queue([maxsize])的简化,只有三个方法------empty(), get(), put()

   class multiprocessing.JoinableQueue([maxsize])是class multiprocessing.Queue([maxsize])的子类,增加了take_done()和join()方法

   注:由于进程之间内存空间不共享,所以必须将实例化后的queue对象当作参数传入其他进程,其他进程才能使用。而且,每传入一次相当于克隆一份,与原来的queue独立,只是python会同步queue中的数据,而不是像在多线程的queue数据只有一份。

 

进程之间的通信:

  multiprocessing.Pipe([duplex])  --------------- 返回表示管道末端的Connection对象(类似与socket中的连接可用于发送和接收数据)的(conn1, conn2)。 

    如果duplex是True(默认值),则管道是双向的。如果duplex是False,则管道是单向的:conn1只能用于接收消息,conn2用于发送消息。

  例二(multiprocessing.Pipe使用演示):

 1 import multiprocessing,time
 2 
 3 class Processing_1(multiprocessing.Process):
 4     def __init__(self, conn):
 5         super(Processing_1, self).__init__()
 6         self.conn = conn
 7     def run(self):
 8         send_data = "this message is from p1"
 9         self.conn.send(send_data)   #使用conn发送数据
10         time.sleep(0.8)
11         recv_data = self.conn.recv() #使用conn接收数据
12         print("p1 recv: " + recv_data)
13         self.conn.close()
14 
15 
16 class Processing_2(multiprocessing.Process):
17     def __init__(self, conn):
18         super(Processing_2, self).__init__()
19         self.conn = conn
20 
21     def run(self):
22         send_data = "this message is from p2"
23         self.conn.send(send_data)
24         time.sleep(0.8)
25         recv_data = self.conn.recv()
26         print("p2 recv: " + recv_data)
27         self.conn.close()
28 
29 if __name__ == "__main__":
30     conn1, conn2 = multiprocessing.Pipe()   #实例化Pipe对象,conn1, conn2分别代表连接两端
31 
32     p1 = Processing_1(conn1)   #将连接对象当作参数传递给子进程
33     p2 = Processing_2(conn2)
34 
35     p1.start()
36     p2.start()
37 
38     p1.join()
39     p2.join()
multiprocessing.Pipe使用演示

 

 

进程之间的数据共享:

  multiprocessing.Manager() ----------- 返回开始的SyncManager对象,可用于在进程之间共享对象。返回的管理器对象对应于生成的子进程,并且具有将创建共享对象并返回相应代理的方法。管理器进程将在垃圾收集或其父进程退出时立即关闭。

  例三(Manager的简单使用):

 1 import multiprocessing,time
 2 import os
 3 
 4 class Processing(multiprocessing.Process):
 5     def __init__(self, d, l):
 6         super(Processing, self).__init__()
 7         self.d = d
 8         self.l = l
 9 
10     def run(self):
11         self.d[os.getpid()] = os.getpid()    #当作正常dict使用即可
12         self.l.append(1)
13         print(self.l)
14 
15 if __name__ == "__main__":
16 
17     manager = multiprocessing.Manager()   #生成Manager 对象
18     d = manager.dict()   #生成共享dict
19     l = manager.list()   #生成共享list
20 
21     p_s = []
22     for i in range(10):
23         p = Processing(d, l)
24         p.start()
25         p_s.append(p)
26 
27     for p in p_s:
28         p.join()
29 
30     print(d)
31     print(l)
Manager简单使用

  manager可以生成以下共享数据对象(常用):

Event ()

Create a shared threading.Event object and return a proxy for it.

Lock ()

Create a shared threading.Lock object and return a proxy for it.

Namespace ()

Create a shared Namespace object and return a proxy for it.

Queue ([maxsize])

Create a shared queue.Queue object and return a proxy for it.

RLock ()

Create a shared threading.RLock object and return a proxy for it.

Semaphore ([value])

Create a shared threading.Semaphore object and return a proxy for it.

Array (typecodesequence)

Create an array and return a proxy for it.

Value (typecodevalue)

Create an object with a writable value attribute and return a proxy for it.

dict ()
dict (mapping)
dict (sequence)

Create a shared dict object and return a proxy for it.

list ()
list (sequence)

Create a shared list object and return a proxy for it.

 

 进程锁:

  进程锁有两种multiprocessing.Lock(非递归锁)和multiprocessing.RLock(递归锁)。

  multiprocessing.Lock(非递归锁):一旦进程或线程获得了锁,随后从任何进程或线程获取它的尝试将阻塞,直到它被释放;任何进程或线程都可以释放它。

  multiprocessing.RLock(递归锁): A recursive lock must be released by the process or thread that acquired it. Once a process or thread has acquired a recursive lock, the same process or thread may acquire it again without blocking; that process or thread must release it once for each time it has been acquired.

  这两种锁都只用两种方法:acquire(block=Truetimeout=None)和release(),它们的使用基本和线程锁类似(只不是要把锁的示例对象当作参数传入其他的进程):http://www.cnblogs.com/God-Li/p/7732407.html

 

进程池:

  为了便于对多进程的管理,通常使用进程池来进行多进程编程(而不是使用multiprocessing.Process)。

  例:

 1 import multiprocessing,os
 2 import time
 3 
 4 
 5 def run():
 6     print(str(os.getpid()) + "-----running")
 7     time.sleep(2)
 8     print(str(os.getpid()) + "-----done")
 9 
10 def done():
11     print("done")
12 
13 def error():
14     print("error")
15 
16 if __name__ == "__main__":
17     pool = multiprocessing.Pool(processes=4)    #实力化进程池对象
18 
19     for i in range(8):
20         # pool.apply(func=run)      #进程池中的进程串行运行
21         pool.apply_async(func=run)
22 
23     pool.close()
24     pool.join()
25     print("finish....")
View Code

  Pool对象常用方法:

apply(func[, args[, kwds]])

Call func with arguments args and keyword arguments kwds. It blocks until the result is ready. Given this blocks, apply_async() is better suited for performing work in parallel. Additionally, func is only executed in one of the workers of the pool.

将任务提交到进程池,只有一个进程在工作,其他进程处于阻塞状态(相当于串行运行)。

apply_async(func[, args[, kwds[, callback[, error_callback]]]])

A variant of the apply() method which returns a result object.

If callback is specified then it should be a callable which accepts a single argument. When the result becomes ready callback is applied to it, that is unless the call failed, in which case the error_callback is applied instead.

If error_callback is specified then it should be a callable which accepts a single argument. If the target function fails, then the error_callback is called with the exception instance.

Callbacks should complete immediately since otherwise the thread which handles the results will get blocked.

将任务提交到进程池,多个进程(进程数量由之前实例化时的processes参数设置)同时运行,callback工作进程完成时(由当前进程的父进程)调用由此传入的任务,error_callback工作进程出错时(由当前进程的父进程)调用由此传入的任务。

close()

Prevents any more tasks from being submitted to the pool. Once all the tasks have been completed the worker processes will exit.

调用此方法后进程池不能在提交新的任务

terminate()

Stops the worker processes immediately without completing outstanding work. When the pool object is garbage collected terminate() will be called immediately.

立即停止工作进程,而不需要等待未完成的工作进程。

join()

Wait for the worker processes to exit. One must call close() or terminate() before using join().

等待进程池中的工作进程结束(在此之前必须调用close()或者terminate())。

注:Pool对象在生成时进程内的进程(阻塞)就已经启动,使用apply(或者apply_async)方法只是将任务提交给线程池,不会再建立新进程。

 

目录
相关文章
|
2天前
|
数据采集 Web App开发 JavaScript
python-selenium模块详解!!!
Selenium 是一个强大的自动化测试工具,支持 Python 调用浏览器进行网页抓取。本文介绍了 Selenium 的安装、基本使用、元素定位、高级操作等内容。主要内容包括:发送请求、加载网页、元素定位、处理 Cookie、无头浏览器设置、页面等待、窗口和 iframe 切换等。通过示例代码帮助读者快速掌握 Selenium 的核心功能。
18 5
|
1天前
|
调度 iOS开发 MacOS
python多进程一文够了!!!
本文介绍了高效编程中的多任务原理及其在Python中的实现。主要内容包括多任务的概念、单核和多核CPU的多任务实现、并发与并行的区别、多任务的实现方式(多进程、多线程、协程等)。详细讲解了进程的概念、使用方法、全局变量在多个子进程中的共享问题、启动大量子进程的方法、进程间通信(队列、字典、列表共享)、生产者消费者模型的实现,以及一个实际案例——抓取斗图网站的图片。通过这些内容,读者可以深入理解多任务编程的原理和实践技巧。
9 1
|
3天前
|
Python
SciPy 教程 之 SciPy 模块列表 13
SciPy教程之SciPy模块列表13:单位类型。常量模块包含多种单位,如公制、二进制(字节)、质量、角度、时间、长度、压强、体积、速度、温度、能量、功率和力学单位。示例代码展示了如何使用`constants`模块获取零摄氏度对应的开尔文值(273.15)和华氏度与摄氏度的转换系数(0.5556)。
8 1
|
1天前
|
Python
SciPy 教程 之 SciPy 模块列表 16
SciPy教程之SciPy模块列表16 - 单位类型。常量模块包含多种单位,如公制、质量、角度、时间、长度、压强、体积、速度、温度、能量、功率和力学单位。示例代码展示了力学单位的使用,如牛顿、磅力和千克力等。
4 0
|
2天前
|
JavaScript Python
SciPy 教程 之 SciPy 模块列表 15
SciPy 教程之 SciPy 模块列表 15 - 功率单位。常量模块包含多种单位,如公制、质量、时间等。功率单位中,1 瓦特定义为 1 焦耳/秒,表示每秒转换或耗散的能量速率。示例代码展示了如何使用 `constants` 模块获取马力值(745.6998715822701)。
6 0
|
2天前
|
JavaScript Python
SciPy 教程 之 SciPy 模块列表 15
SciPy教程之SciPy模块列表15:单位类型。常量模块包含多种单位,如公制、质量、角度、时间、长度、压强、体积、速度、温度、能量、功率和力学单位。功率单位以瓦特(W)表示,1W=1J/s。示例代码展示了如何使用`constants`模块获取马力(hp)的值,结果为745.6998715822701。
7 0
|
3天前
|
Python
SciPy 教程 之 SciPy 模块列表 13
SciPy 教程之 SciPy 模块列表 13 - 单位类型。常量模块包含多种单位:公制、二进制(字节)、质量、角度、时间、长度、压强、体积、速度、温度、能量、功率和力学单位。示例:`constants.zero_Celsius` 返回 273.15 开尔文,`constants.degree_Fahrenheit` 返回 0.5555555555555556。
6 0
|
4月前
|
运维 关系型数据库 MySQL
掌握taskset:优化你的Linux进程,提升系统性能
在多核处理器成为现代计算标准的今天,运维人员和性能调优人员面临着如何有效利用这些处理能力的挑战。优化进程运行的位置不仅可以提高性能,还能更好地管理和分配系统资源。 其中,taskset命令是一个强大的工具,它允许管理员将进程绑定到特定的CPU核心,减少上下文切换的开销,从而提升整体效率。
掌握taskset:优化你的Linux进程,提升系统性能
|
4月前
|
弹性计算 Linux 区块链
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
137 4
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
|
3月前
|
算法 Linux 调度
探索进程调度:Linux内核中的完全公平调度器
【8月更文挑战第2天】在操作系统的心脏——内核中,进程调度算法扮演着至关重要的角色。本文将深入探讨Linux内核中的完全公平调度器(Completely Fair Scheduler, CFS),一个旨在提供公平时间分配给所有进程的调度器。我们将通过代码示例,理解CFS如何管理运行队列、选择下一个运行进程以及如何对实时负载进行响应。文章将揭示CFS的设计哲学,并展示其如何在现代多任务计算环境中实现高效的资源分配。