python multiprocessing模块
python multiprocessing模块
multiprocessing
multiprocessing支持子进程、通信和共享数据、执行不同形式的同步,提供了Process、Queue、Pipe、Lock等组件。 创建进程的类:Process([group[, target[, name[, args[, kwargs]]]]]) target表示调用对象 args表示调用对象的位置参数元组。 kwargs表示调用对象的字典。name为别名。 group表示线程组。 方法: is_alive():返回进程是否 join([timeout])运行:阻塞当前上下文环境的进程,直到调用此方法的进程终止或到达指定timeout(可选参数) run():start()调用run方法,如果实例进程时未制定target,这start执行 默认run()方法 start():进程准备就绪,等待CPU调度 terminate():不管任务是否完成,立即停止工作进程 其中,Process以start()启动某个进程。 属性:authkey、daemon(要通过start() 设置)、exitcode(进程在运行时为None、如果为–N,表示被信号N结束)、name、pid。其中daemon是父进程终止后自动终止,且自己不能产生新进程,必须在start() 之前设置。
Process类
构造方法: Process([group [, target [, name [, args [, kwargs]]]]]) group: 线程组,目前还没有实现,库引用中提示必须是None; target: 要执行的方法; name: 进程名; args/kwargs: 要传入方法的参数。 实例方法: is_alive():返回进程是否在运行。 join([timeout]):阻塞当前上下文环境的进程程,直到调用此方法的进程终止或到达指定的timeout(可选参数)。 start():进程准备就绪,等待CPU调度 run():strat()调用run方法,如果实例进程时未制定传入target,这star执行t默认run()方法。 terminate():不管任务是否完成,立即停止工作进程 属性: daemon:和线程的setDeamon功能一样 name:进程名字。 pid:进程号。
进程的调用
关于创建多线程
四种方法可以创建多线程 (1)系统初始化:启动操作系统时开启的线程,比如前后台进程。 (2)执行了正在运行的进程所调用的进程创建系统进程:一个正在运行的进程经常发出系统调用,以便创建一个或多个新进程协助其工作。 (3)用户请求创建一个新进程:双击图标,打开一个新程序,又比如运行我们编写的程序等。 (4)一个批处理作业的初始化:这种只有在大型机的批处理系统中应用,在这里不提及。 注意:所有情形中,新进程都是由一个已存在的进程执行一个用于创建进程的系统调用而创建的,这个进程所做的工作是,执行一个用来创建新进程的系统调用,系统调用会通知操作系统创建一个新进程,并且直接或间接地指定在该进程中运行的程序。 UNIX:fork,这个系统调用会创建一个与调用进程相同的副本,父进程与子进程拥有相同的存储映像,通常,子进程接着执行execve或一个类似的系统调用,以修改其存储映像并运行一个新的程序。 Windows:Win32函数调用Create'Process创建进程,也负责进行父子进程的复制,该调用由10个参数。 UNIX、Windows中,进程创建之后,父子进程有各自不同的地址,如果其中某个进程在其地址空间进行修改(可写内存),这个修改对于其他进程而言是不可见的。(不共享内存) UNIX,子进程的初始地址是父进程的副本,但这里涉及两个不同的地址看见,不可写的内存是共享的(某些UXIX的实现使程序正文在两者共享,因为它不能被修改),对于新创建的进程而言,有可能共享其创建者的其他资源,比如打开的文件等。 Windows中,从一开始父进程与子进程的地址空间就不一样。
#创建调用多进程 #函数 # import multiprocessing # import time # # def worker_1(interval): # print("worker_1") # time.sleep(interval) # print("end worker_1") # # def worker_2(interval): # print("worker_2") # time.sleep(interval) # print("end worker_2") # # # if __name__ == "__main__": # p1 = multiprocessing.Process(target = worker_1, args = (2,)) # p2 = multiprocessing.Process(target = worker_2, args = (3,)) # p1.start() # p2.start() # p1.join() # p2.join() # print('finsh end') #定义成类 # import multiprocessing # import time # # class ClockProcess(multiprocessing.Process): # def __init__(self, interval): # multiprocessing.Process.__init__(self) # self.interval = interval # # def run(self): # n = 5 # while n > 0: # print("the time is {0}".format(time.ctime())) # time.sleep(self.interval) # n -= 1 # # if __name__ == '__main__': # p = ClockProcess(3) # p.start()
进程同步
注意:这里使用锁需要把锁传递进函数,因为是使用的是不同的进程,这里有复制拷贝!!!
from multiprocessing import Process, Lock def f(l, i): with l.acquire(): print('hello world %s'%i) if __name__ == '__main__': lock = Lock() for num in range(10): Process(target=f, args=(lock, num)).start()
进程间通讯
进程对列Queue
from multiprocessing import Process, Queue import queue def f(q,n): #q.put([123, 456, 'hello']) q.put(n*n+1) print("son process",id(q)) if __name__ == '__main__': q = Queue() #try: q=queue.Queue() print("main process",id(q)) for i in range(3): p = Process(target=f, args=(q,i)) p.start() print(q.get()) print(q.get()) print(q.get())
管道
The Pipe()
function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:
from multiprocessing import Process, Pipe def f(conn): conn.send([12, {"name":"yuan"}, 'hello']) response=conn.recv() print("response",response) conn.close() print("q_ID2:",id(child_conn)) if __name__ == '__main__': parent_conn, child_conn = Pipe() print("q_ID1:",id(child_conn)) p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) # prints "[42, None, 'hello']" parent_conn.send("儿子你好!") p.join()
The two connection objects returned by Pipe()
represent the two ends of the pipe. Each connection object has send()
and recv()
methods (among others). Note that data in a pipe may become corrupted if two processes (or threads) try to read from or write to the same end of the pipe at the same time. Of course there is no risk of corruption from processes using different ends of the pipe at the same time.
Managers
Queue和pipe只是实现了数据交互,并没实现数据共享,即一个进程去更改另一个进程的数据。
A manager object returned by Manager()
controls a server process which holds Python objects and allows other processes to manipulate them using proxies.
A manager returned by Manager()
will support types list
, dict
, Namespace
, Lock
, RLock
, Semaphore
, BoundedSemaphore
, Condition
, Event
, Barrier
, Queue
, Value
and Array
. For example:
from multiprocessing import Process, Manager def f(d, l,n): d[n] = '1' d['2'] = 2 d[0.25] = None l.append(n) #print(l) print("son process:",id(d),id(l)) if __name__ == '__main__': with Manager() as manager: d = manager.dict() l = manager.list(range(5)) print("main process:",id(d),id(l)) p_list = [] for i in range(10): p = Process(target=f, args=(d,l,i)) p.start() p_list.append(p) for res in p_list: res.join() print(d) print(l)
有一点需要注意,在共享 list 时,像下面这样写 f函数 是不起作用的。
def f(my_list, my_dict): my_list = [10, 11] my_dict['a'] = 1 my_dict['b'] = 2
这样写相当于重新定义了一个局部变量,并没有作用到原来的 list 上,必须使用 append,extend 等方法。
进程池
进程池内部维护一个进程序列,当使用时,则去进程池中获取一个进程,如果进程池序列中没有可供使用的进进程,那么程序就会等待,直到进程池中有可用进程为止。
进程池中有两个方法:
- apply
- apply_async
from multiprocessing import Process,Pool import time,os def Foo(i): time.sleep(1) print(i) return i+100 def Bar(arg): print(os.getpid()) print(os.getppid()) print('logger:',arg) pool = Pool(5) Bar(1) print("----------------") for i in range(10): #pool.apply(func=Foo, args=(i,)) #pool.apply_async(func=Foo, args=(i,)) pool.apply_async(func=Foo, args=(i,),callback=Bar) pool.close() pool.join() print('end')
相关文章
- Python-Excel 模块哪家强 #华为云·寻找黑马程序员#
- python中的随机函数random的用法示例
- python paramiko模块学习分享
- 如何编写一个python项目
- [Python] Histograms for analysis Daily return
- python远程登录服务器(paramiko模块安装和使用)
- 小白学 Python 数据分析(18):Matplotlib(三)常用图表(上)
- 小白学 Python 爬虫(27):自动化测试框架 Selenium 从入门到放弃(上)
- Python的Asyncore异步Socket模块及实现端口转发的例子
- python使用cPickle模块序列化实例
- python内建集合模块collections功能,计数,有序,双向队列
- [转]Python 命令行参数和getopt模块详解
- paip.python连接mysql最佳实践o4
- Python编程语言学习:python语言中快速查询python自带模块&函数的用法及其属性方法、如何查询某个函数&关键词的用法、输出一个类或者实例化对象的所有属性和方法名之详细攻略
- Python编程语言学习:包导入和模块搜索路径(包路径)简介、使用方法(python系统环境路径的查询与添加)之详细攻略
- Python语言编程学习:文件路径变量修改,利用os模块固定文件父路径,变换文件子路径实现代码
- 100天精通Python(进阶篇)——第38天:多进程multiprocessing模块基础+代码实战
- 已解决Python正确安装使用hanlp模块分词器
- 【Python自动化办公】分享几个好用到爆的模块,建议收藏
- 【项目实战】Python实现深度神经网络RNN-LSTM分类模型(医学疾病诊断)
- 命令行启动模块的Python代码研究
- Python 协程模块 asyncio 使用指南
- linux的shell脚本运行python程序
- 【华为机试真题 Python实现】子串分离
- Python:使用 venv 来创建虚拟环境
- Python编程:twine模块打包python项目上传pypi
- Python:mysql-connector-python模块对MySQL数据库进行增删改查
- python基础===Python 迭代器模块 itertools 简介
- 多版本Python共存时pip给指定版本的python安装package的方法