动图从哪个网站做,建立网站要钱吗,清风WordPress,如何做网站规划1、为什么要掌握进程间通信 Python代码效率由于受制于GIL全局锁限制#xff0c;多线程不能利用多核CPU来加速#xff0c;而多进程方式却可以绕过GIL限制, 发挥多CPU加速的优势#xff0c;达到提高程序的性能的目的。 然而进程间通信却是不得不考虑的问题。 进程不同于线程多线程不能利用多核CPU来加速而多进程方式却可以绕过GIL限制, 发挥多CPU加速的优势达到提高程序的性能的目的。 然而进程间通信却是不得不考虑的问题。 进程不同于线程进程有自己的独立内存空间不能使用全局变量在进程间传递数据。 实际项目需求中常常存在密集计算、或实时性任务进程之间有时需要传递大量数据如图片、大对象等传递数据如果通过文件序列化、或网络接口来进行难以满足实时性要求采用redis或者kaffka, rabbitMQ 之第3方消息队列包又使系统复杂化了而且多次重复编码解码数据效率并不高。 Python multiprocessing 模块本身就提供了丰富高效的进程间通信机制包括消息机制、同步机制、共享内存等。 了解并掌握 python 各种进程间通信的特点、使用方式以及安全机制可以帮助软件工程师在实际项目选用最合适的通信方式从而获得程序运行的最佳性能。 2、进程间各类通信方式简介
下面列出了python支持的各种进程间通信方式 各类通信方式的适用场景 主要是由于不同应用场景以及对性能的不同需求使用通信方式。 多进程同时访问数据库读写文件等场景需要添加同步机制可采用Lock, Event, Semaphore等机制。而Queue队列、Pipe 适合大多数场景如 生产者 – 消息者订阅 – 发布Director – Worker等场景在大多数情况下Queue Pipe即可满足进程间通信的需求。由于Queue对入列、出列对象需要进行序列化操作等影响了效率。如果对性能要求高可采用共享变量方式交换数据。如果希望python性能飞起来共享内存方式是不二之选非常适合进程之间交换图片、视频等大块数据只是对开发者要求更高必须熟悉bytes, bytearray, memoryview类型的使用以及pickle, json序列化、字符串编解码等知识。 关于进程间通信的内存安全 同1个应用程序的多进程间通信可能会因同抢意外销毁等原因造成共享变量异常或者使用不当造成内存占用过高也可能被其它进程非授权访问。 Multiprocessing 模块提供的Queue, Pipe, Lock, Event 对象都已实现了进程间通信安全机制。 需要注意采用共享内存方式通信需要在代码中开发者自己来跟踪、销毁这些共享内存变量。除非开发者很清楚共享内存使用特点否则不建议直接使用此共享内存而是通过Manager管理器来使用共享内存。
进程管理器Manager Multiprocessing提供了管理器Manager类SharedMemoryManager类可解决进程通信的内存安全问题可以将各种共享数据加入管理器包括 list, dict, Queue, Lock, Event, SharedMemory 等由其统一跟踪与销毁。
下面按类别介绍种通信方式的特点以及代码开发步骤。
3、消息机制通信
1) 管道 Pipe 通信方式
类似于1上简单的socket通道双端均可收发消息。 Pipe 对象的构建方法
parent_conn, child_conn Pipe(duplexTrue/False) 参数说明 duplexTrue, 管道为双向通信duplexFalse, 管道为单向通信只有child_conn可以发消息parent_conn只能接收。 示例代码
from multiprocessing import Process, Pipe
def myfunction(conn):conn.send([hi!! I am Python])conn.close()if __name__ __main__:parent_conn, child_conn Pipe()p Process(targetmyfunction, args(child_conn,))p.start()print (parent_conn.recv() )p.join()#结果
#[hi!! I am Python]
2) 消息队列Queue通信方式
Multiprocessing 的Queue 类是在python queue 3.0版本上修改的 可以很容易实现生产者 – 消息者间传递数据而且Multiprocessing的Queue 模块实现了lock安全机制。 Queue模块共提供了3种类型的队列。
(1) FIFO queue , 先进先出
class queue.Queue(maxsize0)(2) LIFO queue, 后进先出 实际上就是堆栈
class queue.LifoQueue(maxsize0)(3) 带优先级队列 优先级最低entry value lowest 先了列
class queue.PriorityQueue(maxsize0)Multiprocessing.Queue类的主要方法
methodDescriptionqueue.qsize()返回队列长度queue.full()队列满返回 True, 否则返回Falsequeue.empty()队列空返回 True, 否则返回Falsequeue.put(item)将数据写入队列queue.get()将数据抛出队列 queue.put_nowait(item), queue.get_nowait()无等待写入或抛出
说明
put(), get() 是阻塞方法 而put_notwait(), get_nowait()是非阻塞方法。Multiprocessing 的Queue类没有提供Task_done, join方法
Queue模块的其它队列类
(1) SimpleQueue 简洁版的FIFO队列, 适事简单场景使用
(2) JoinableQueue子类 Python 3.5 后新增的 Queue的子类拥有 task_done(), join() 方法
task_done()表示最近读出的1个任务已经完成。join()阻塞队列直到queue中的所有任务都已完成。
producer – consumer 场景使用Queue的示例
import multiprocessing
import random
def producer(numbers, q):for x in numbers:if x % 2 0:if q.full():print(queue is full)breakq.put(x)print(fput {x} in queue by producer)return Nonedef consumer(q):while not q.empty():print(ftake data {q.get()} from queue by consumer)return Noneif __name__ __main__:# 设置1个queue对象最大长度为5qu multiprocessing.Queue(maxsize5,)# 创建producer子进程把queue做为其中1个参数传给它该进程负责写p5 multiprocessing.Process(nameproducer-1,targetproducer,args([random.randint(1, 100) for i in range(0, 10)], qu))p5.start()p5.join()#创建consumer子进程把queue做为1个参数传给它该进程中队列中读p6 multiprocessing.Process(nameconsumer-1,targetconsumer,args(qu,))p6.start()p6.join()print(qu.qsize())#结果
# put 80 in queue by producer
# put 32 in queue by producer
# take data 80 from queue by consumer
# take data 32 from queue by consumer
# 0
4、同步机制通信
(1) 同步锁 – Lock Multiprocessing也提供了与threading 类似的同步锁机制确保某个时刻只有1个子进程可以访问某个资源或执行某项任务, 以避免同抢。 例如多个子进程同时访问数据库表时如果没有同步锁用户A修改1条数据后还未提交此时用户B也进行了修改可以预见用户A提交的将是B个修改的数据。 添加了同步锁可以确保同时只有1个子进程能够进行写入数据库与提交操作。 如下面的示例同时只有1个进程可以执行打印操作。 from multiprocessing import Process, Lockdef f(l, i):l.acquire()try:print(hello world, i)finally:l.release()if __name__ __main__:lock Lock()for num in range(10):Process(targetf, args(lock, num)).start()# 结果
# hello world 4
# hello world 0
# hello world 5
# hello world 7
# hello world 2
# hello world 3
# hello world 8
# hello world 9
# hello world 6
# hello world 1
不过acquire() 未能获取锁的话会阻塞当前线程直到其它线程将锁release掉才会继续执行。若有两个以上的锁对象容易造成死锁多人项目运用Lock通信时应慎重。
(2) 子进程间协调机制 – Event
Event 机制的工作原理
1个event 对象实例管理着1个 flag标记, 可以用set()方法将其置为true, 用clear()方法将其置为false, 使用wait()将阻塞当前子进程直至flag被置为true. 这样由1个进程通过event flag 就可以控制、协调各子进程运行。 Event 是一种进程间通信的信号机制以异步方式运行。此对象与pytho signal模块的中断信号工作机制是不同的。 Event 对象的使用方法
1)主进程 创建1个event 对象如 flag multiprocessing.Event() , 做为参数传给各子进程 2) 子进程joo_b: 不受event影响,通过event 控制其它进程的运行
先clear()将event 置为False, 占用运行权.完成工作后用set()把flag置为True。
3)子进程 joo_a: 受 event 影响
event对象flag设置为 wait() 状态暂停运行直到flag重新变为True恢复运行
Event对象主要方法
set(), clear()设置 True/False, wait() 使进程等待直到flag被改为true.is_set() Return True if and only if the internal flag is true.
详细代码如下
import multiprocessing
import time
import randomdef joo_a(q, ev):print(subprocess joo_a start)if not ev.is_set():ev.wait()q.put(random.randint(1, 100))print(subprocess joo_a ended)def joo_b(q, ev):print(subprocess joo_b start)ev.clear()time.sleep(2)q.put(random.randint(200, 300))ev.set()print(subprocess joo_b ended)def main_event():qu multiprocessing.Queue()ev multiprocessing.Event()sub_a multiprocessing.Process(targetjoo_a, args(qu, ev))sub_b multiprocessing.Process(targetjoo_b, args(qu, ev,))sub_a.start()sub_b.start()# ev.set()sub_a.join()sub_b.join()while not qu.empty():print(qu.get())if __name__ __main__:main_event()# 结果
# subprocess joo_a start
# subprocess joo_b start
# subprocess joo_b ended
# subprocess joo_a ended
# 267
# 595、共享变量 进程间通信不能使用全局变量由于每个子进程运行在自己独立内存空间只是创建时全局变量复制过来了。因此在子进程A中修改了全局变量的值子进程B是无法看到的。 原理 在主进程中创建ctype共享变量对象再被子进程继承 。 优点 速度比queue快很多
Python相关支持模块 • multiprocessing.Value multiprocessing.Array. 是Sharedctypes的派生类 • multiprocessing.sharedctypes, 则支持更全面ctypes数据类型.
Value()语法:
Value(typecode_or_type, *args, lockTrue)参数说明
typecode: ctypes 类型缩写,或 ctype类型名 如’i’ 表示 c_init, ‘d’ 表示 c_double, ‘c’ 表示 c_char等。*args 传入值Lock locktrue, 表示添加1个锁
返回值 返回一个从共享内存上创建的 ctypes 对象 示例
camera_unit multiprocessing.Value(ctypes.c_int,0)
servo_on_flag multiprocessing.Value(ctypes.c_bool,False)
target_angle multiprocessing.Value(‘d’,30.0) 注意 使用 share memory 要考虑同抢等问题释放等问题需要手工实现。因此在使用共享变量时建议使用Manager管程来管理这些共享变量。
import multiprocessing
def func(num):num.value 10.78 # 子进程改变数值的值主进程跟着改变if __name__ __main__:num multiprocessing.Value(d, 10.0)
# d表示数值,主进程与子进程可共享这个变量。p multiprocessing.Process(targetfunc, args(num,))p.start()p.join()print(num.value)# 结果
# 10.78
进程之间共享数据(数组型)
multiprocessing.Array() 返回1个array, 元素为ctypes类型
import multiprocessingdef func(num):num[2] 9999 # 子进程改变数组主进程跟着改变if __name__ __main__:num multiprocessing.Array(i, [1, 2, 3, 4, 5])p multiprocessing.Process(targetfunc, args(num,))p.start()p.join()print(num[:])# 结果
# [1, 2, 9999, 4, 5]
sharedctypes.RawValue() 使用方式与 Value() 是类似的只是typecode必须是ctypes类型全称本文不再介绍。
6、共享内存 Shared_memory Python 提供了一个 SharedMemory 类用于分配和管理多核或对称多处理器SMP机器上进程间的共享内存而非直接共享变量 共享内存方式与 ctypes共享变量是不同的。前者共享的是二进制内存后者共享对象为ctype变量更友好。 ShareMemory处理对象为binary类型因此对编程者并不友好但优点是处理速度快。 注意直接使用SharedMemory 也存在着同抢、泄露隐患应通过SharedMemory Manager 管理器来使用, 以确保内存安全。 1 SharedMemory对象编程步骤
创建共享内存区
multiprocessing.shared_memory.SharedMemory(namenone, createFalse, size0)参数说明
name: 全局唯一的名称create 指定创建一个新的共享内存块 (True) 还是连接到已存在的共享内存块(False)
使用方法 父进程创建shared_memory 后子进程可以使用它当不再需要后使用close(), 删除使用unlink()方法相关属性 获取内存区内容 shm.buf为 memoryview 类型 获取内存区名称 shm.name 获取内存区字节数: shm.size
SharedMemory 编程步骤1 在主进程中创建新的SharedMemory 块
shm shared_memory.SharedMemory(nameshmArray,createTrue, size10)shmArray为共享内存的名字其它进程中可以使用它来attach 初始化
shm.buf[0:5] bytearray([10,3,8,1,5]) 2将共享内存对象做为参数传给子进程任务函数
p1 mp.Process(targettask_a,args(shm,)) 3 子进程读取修改共享内存对象 buff shm.bufbuff[5] 200 # modify the shared memoryprint(fshared_memeory: {buff[0:9].tolist()})4 使用attach 方式共享内存对象。 这种方式工不需将共享内存对象做为参数传给子进程中。子进程通过attach方式连接到主进程已创建好的共享内存对象。 具体使用初始化SharedMemory对象时name为已存在共享内存对象名create参数设置为False, 如下
shm_b shared_memory.SharedMemory(nameshmArray,createFalse)这样shm_b指向了主进程中创建的shm.5 释放共享内存对象 子进程通过attach方式访问使用用close()方法关闭。所有子进程都不再使用后调用unlink()方法释放共享内存对象。
示例 创建两个进程使用共享内存通信. 进程p1接受传入SharedMemory 对象进程p2使用attach 方式访问主进程创建的SharedMemory 对象。
from multiprocessing import shared_memory
import multiprocessing as mp
import osdef task_a(shm):Task function for multiprocessingInputs:shm: shared_memory object# create a shared memory object, attached to exsisting one.buff shm.buf# modify the shared memorybuff[5] 200print(fcurrent process: { os.getpid()}, shared_memeory: {buff[0:9].tolist()})return Nonedef task_b():Task function for multiprocessingno input, access existing sharedMemory object.# create shared memory object, attached to exsisting one.shm_b shared_memory.SharedMemory(nameshmArray,createFalse)buff shm_b.bufbuff[6] 100# Pytho SharedMemory class may have a bugthe length of attached SharedMemory object# is not equal to original one.print(fcurrent process: { os.getpid()}, shared_memeory: {buff[0:9].tolist()})print(fbuff size: {shm_b.size})shm_b.close()return Noneif __name____main__:# 创建两个进程使用共享内存通信# create a shared memory objectshm shared_memory.SharedMemory(nameshmArray,createTrue, size10)shm.buf[0:5] bytearray([10,3,8,1,5]) # initialize the shared memoryp1 mp.Process(targettask_a,args(shm,)) # 将sharedMmemory对象作为参数传递给子进程p2 mp.Process(targettask_b,) # 无参数p1.start()p1.join()p2.start()p2.join()print(in main process, shared_memory: ,shm.buf.tolist() )shm.close()shm.unlink()# 结果
# current process: 17732, shared_memeory: [10, 3, 8, 1, 5, 200, 0, 0, 0]
# current process: 12356, shared_memeory: [10, 3, 8, 1, 5, 200, 100, 0, 0]
# buff size: 4096
# in main process, shared_memory: [10, 3, 8, 1, 5, 200, 100, 0, 0, 0]
多个子进程对同1个SharedMemory对象同时修改操作如何避免同抢 参考第4章子进程中在操作之前使用 Lock同步锁来避免同抢。
2 ShareableList 共享列表 sharedMemory类还提供了1个共享列表类型这样就更方便了进程间可以直接共享python强大的列表 构建方法 multiprocessing.shared_memory.ShareableList(sequenceNone, *, nameNone) from multiprocessing import shared_memorya shared_memory.ShareableList([howdy, bHoWdY, -273.154, 100, None, True, 42])[ type(entry) for entry in a ]
[class str, class bytes, class float, class int, class NoneType, class bool, class int]a[2]
-273.154a[2] -78.5a[2]
-78.5a[2] dry ice # Changing data types is supported as wella[2]
dry icea[2] larger than previously allocated storage space
Traceback (most recent call last):...
ValueError: exceeds available storage for existing stra[2]
dry icelen(a)
7a.index(42)
6a.count(bhowdy)
0a.count(bHoWdY)
1a.shm.close()a.shm.unlink()del a # Use of a ShareableList after call to unlink() is unsupportedb shared_memory.ShareableList(range(5)) # In a first processc shared_memory.ShareableList(nameb.shm.name) # In a second processc
ShareableList([0, 1, 2, 3, 4], name...)c[-1] -999b[-1]
-999b.shm.close()c.shm.close()c.shm.unlink()7、管理器Manager
Multiprocessing 提供了 Manager 管理器类当调用1个Manager实例对象的start()方法时会创建1个manager进程其唯一目的就是管理共享变量、共享内存, 避免出现进程间共享数据不同步内存泄露等现象。
其原理如下 Manager管理器相当于提供了进间程共享内存对象的服务不仅可以被主进程创建的多个子进程使用还可以被其它进程访问甚至跨网络访问。本文仅聚焦于由单一主进程创建的各子进程之间的通信。
1 Manager的主要数据结构
相关类multiprocessing.Manager 还有两个子类
multiprocessing.managers.SharedMemoryManagermultiprocessing.managers.BaseManager
支持共享变量类型
Manger类支持的共享变量类型有 dic, listQueue, Lock, Event, Condition, Semaphore, Barrier等类型 通过代理可支持python各种类型 以及ctypes 类型 SharedMemoryManager子类额外支持 SharedMemoryShareableList类型。
2 使用步骤
1创建管理器对象 snm Manager() 或者 snm SharedMemoryManager()
2创建共享内存变量 新建list, dict sl snm.list() sd snm.dict()
新建1块bytes共享内存变量需要指定大小 sx snm.SharedMemory(size)
新建1个共享列表变量可用列表来初始化 sl snm.ShareableList(sequence) 如 sl smm.ShareableList([‘howdy’, b’HoWdY’, -273.154, 100, True])
新建1个queue, 使用multiprocessing 的Queue类型 q snm.Queue()
传入子进程
def foo(s, d):.....p1 Process( targetfoo, args(sl, sd, ) )
p1.start()
p1.join()示例
from multiprocessing import Process, Managerdef f(d, l):d[1] 1d[2] 2d[0.25] Nonel.reverse()if __name__ __main__:with Manager() as manager:d manager.dict()l manager.list(range(10))p Process(targetf, args(d, l))p.start()p.join()print(d)print(l)将打印
{0.25: None, 1: 1, 2: 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]3 销毁共享内存变量
方法一 调用snm.shutdown()方法会自动调用每个内存块的unlink()方法释放内存。或者 snm.close()方法二 使用with语句结束后会自动释放所有manager变量 with SharedMemoryManager() as smm:
... sl smm.ShareableList(range(2000))
... # Divide the work among two processes, storing partial results in sl
... p1 Process(targetdo_work, args(sl, 0, 1000))
... p2 Process(targetdo_work, args(sl, 1000, 2000))
... p1.start()
... p2.start() # A multiprocessing.Pool might be more efficient
... p1.join()
... p2.join() # Wait for all work to complete in both processes
... total_result sum(sl) # Consolidate the partial results now in sl4 向管理器注册自定义类型
managers的子类BaseManager提供register()方法支持注册自定义数据类型。如下例注册1个自定义MathsClass类并生成实例。
from multiprocessing.managers import BaseManagerclass MathsClass:def add(self, x, y):return x ydef mul(self, x, y):return x * yclass MyManager(BaseManager):passMyManager.register(Maths, MathsClass)if __name__ __main__:with MyManager() as manager:maths manager.Maths()print(maths.add(4, 3)) # prints 7print(maths.mul(7, 8)) 7、总结 Python多进程(multiprocessing) 编程是绕开GIL提升程序性能的重要方式进程间通信方式包括消息机制pipe, queue)、同步机制( Lock, Event) 、Shared MemoryValue, Array, Shared_Memory, etc)等。 直接使用Shared Memory共享内存是不安全的Multiprocessing.Manager模块提供了安全管理共享内存变量的管理器功能。 在实际编程时根据主进程与子进程子进程之间所要交换数据的类型、大小频度、实时性等需求来选择适合的通信方式。 8、进程、线程对比
8.1功能 进程能够完成多任务比如在一条电脑上能够同时运行多个QQ 线程能够完成多任务比如一个QQ中多个聊天窗口 定义的不同线程是系统进行资源分配和调度的一个独立单位 线程是进程的一个实体是CPU调度和分配的基本单位它是比进程更小的能独立运行的基本单位线程自己基本上不拥有系统资源只拥有一点在运行中必不可少的资源如程序计数器一组寄存器和栈但是它可与同一个进程的其他线程共享进程所拥有的全部资源 8.2 区别 一个程序至少有一个进程一个进程至少有一个线程 线程的划分尺度小于进程资源比进程少使得多线程程序的并发性高 进程在执行过程中拥有独立的内存单元而多个线程共享内存从而极大地提高了程序的运行效率 线程不能独立执行必须依存在进程中 可以将进程理解为工厂中的一条流水线而其中的线程就是流水线中的工人 8.3 优缺点 线程线程执行开销小但不利于资源的管理和保护 进程进程执行开销大但利于资源的管理和保护 参考
1.由浅入深掌握各种 Python 进程间通信方式(建议收藏
2.一文读懂Python进程间通信的几种方式
