一文速通Python并行计算：10 Python多进程编程-进程之间的数据共享-基于共 ...

写过一篇 · 3 天前

一文速通 Python 并行计算：10 Python 多进程编程-进程之间的数据共享-基于共享内存和数据管理器

摘要：

Python 多进程通信中，共享内存通过 Value 和 Array 实现高效数据共享，直接操纵物理内存避免拷贝，需同步锁保证安全，支持数值、数组及自界说布局体（借助 ctypes 模块）。Manager 则通过服务进程管理共享对象（如 dict、list），支持更多数据范例但性能较低。共享内存适合高频小数据，Manager 适用于复杂数据布局，两者均需进程同步机制确保数据一致性，兼顾服从与灵活性。

关于我们更多介绍可以查看云文档：Freak 嵌入式工作室云文档，或者访问我们的 wiki：****https://github.com/leezisheng/Doc/wik

正文

1. 共享内存

共享内存，顾名思义就是允许两个不相关的进程访问同一个逻辑内存，共享内存是两个正在运行的进程之间共享和传递数据的一种非常有用的方式。差别进程之间共享的内存通常为同一段物理内存。进程可以将同一段物理内存连接到他们本身的地址空间中，所有的进程都可以访问共享内存中的地址。假如某个进程向共享内存写入数据，所做的改动将立即影响到可以访问同一段共享内存的任何其他进程。

其特点包括：

共享内存是一种最为高效的进程间通信方式，进程可以直接读写内存，而不需要任何数据的拷贝。
为了在多个进程间交换信息，内核专门留出了一块内存区，可以由需要访问的进程将其映射到本身的私有地址空间。进程就可以直接读写这一块内存而不需要举行数据的拷贝，从而大大进步服从。（文件映射）
由于多个进程共享一段内存，因此也需要依靠某种同步机制。

虽然共享内存可以实现快速在进程间传递数据，但也存在内存中的内容会被其他进程覆盖或者窜改的缺点，需要用进程间同步方式保证数据安全。
multiprocessing 中提供了 Value 和 Array，可以将数据存储在共享内存映射中，供全部进程访问。
1.1 共享内存-数值型

起首我们来看，value 的数据格式：

Value(typecode_or_type, *args, lock=True)

复制代码

功能：

得到一个共享内存对象，而且存入初始值，method of multiprocessing
返回 Returns a synchronized shared object（同步共享对象）
typecode_or_type：界说了返回范例（转换成 C 语言中存储范例），它要么是一个 ctypes 范例，要么是一个代表 ctypes 范例的 code。ctypes 是 python 的一个外部函数库，它提供了和 C 语言兼容的数据范例，可以调用 DLLs 或共享库的函数，能被用作在 python 中包裹这些库。
*args：开辟一个空间，并赋一个 args 值，值得范例不限
lock 默认自动上锁(同步锁)

示例代码如下，可以看到子进程改变数值的值，主进程跟着改变：

import multiprocessing
def func(num):
num.value = 11
if __name__ == "__main__":
# i表示整型,主进程与子进程共享这个value。
num = multiprocessing.Value("i", 10)
print(num.value)
p = multiprocessing.Process(target=func, args=(num,))
p.start()
p.join()
print(num.value)

复制代码

1.2 共享内存-数组型

Array 的数据格式：

Array(typecode_or_type, size_or_initializer, *, lock=True)

复制代码

功能：

利用基本类似于 Value，Returns a synchronized shared array （是一个可迭代对象）
typecode_or_type：界说转换成 C 语言的存储范例；
size_or_initializer：初始化共享内存空间，假如是一个整数，那就会当做数组的长度，而且整个数组的内存会初始化为 0。否则，会被当成一个序列用于初始化数组中的每一个元素。
lock 默认自动上锁(同步锁)

Value、Array 数据存储范例参数如下（部分）：

示例代码如下，可以看到子进程改变数列元素，主进程跟着改变：

import multiprocessing
def func(num):
num[2] = 9999 # 子进程改变数组，主进程跟着改变
if __name__ == "__main__":
num = multiprocessing.Array("i", [1, 2, 3, 4, 5]) # 主进程与子进程共享这个数组
print(num[:])
p = multiprocessing.Process(target=func, args=(num,))
p.start()
p.join()
print(num[:])

复制代码

1.3 自界说数据布局

这里利用 ctypes 中类来界说布局体，用于界说布局体的类需要继承自 ctypes 的 Structure 基类，而后通过界说类的 _fields_ 属性来界说布局体的构成。_fields_ 属性一般界说为一个二维的 tuple，而对于其中的每一个一维 tuple，其需要界说两个值，第一个值为一个字符串，用作布局体内部的变量名，第二个值为一个 ctypes 范例，用于界说当前布局体变量所界说的数据范例。留意，在 Python 中界说的布局体，其变量名,类名等均可以差别于 C 语言中的变量名，但布局体变量的数量、数据范例与次序必须严酷对应于 C 源码中的界说，否则可能将导致内存访问出错。
代码如下：

**from multiprocessing import Process, Lock**
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_double
_# 自定义数据结构_
class Point(Structure):
_# 定义了一个结构体类型，其等同于C中的struct声明。_
_# 此结构体定义了两个结构体变量：x对应于一个int类型，y对应于一个double类型。_
_fields_ = [('x', c_double), ('y', c_double)]
def modify(n, x, s, A):
n.value **= 2
x.value **= 2
s.value = s.value.upper()
for a in A:
a.x **= 2
a.y **= 2
if __name__ == '__main__':
lock = Lock()
n = Value('i', 7) _# 定义int型_
x = Value(c_double, 1.0/3.0, lock=False) _# 定义ctypes类型_
s = Array('c', b'hello world', lock=lock) _# 定义字符串_
A = Array(Point, [(1.875, -6.25), (-5.75, 2.0), (2.375, 9.5)], lock=lock)
_# 定义Point类型的数组_
p = Process(target=modify, args=(n, x, s, A))
p.start()
p.join()
print(n.value)
print(x.value)
print(s.value)
print([(a.x, a.y) for a in A])

复制代码

2.数据管理器

除了共享内存的方式，另一种进程间共享对象的方式是利用服务进程：
由 Manager() 返回的管理器对象控制一个服务进程，该进程保存 Python 对象并允许其他进程利用代理操纵它们，（留意，Manager 需要手动上锁）。
管理者有以下特性：

它控制着管理共享对象的服务进程；
它确保当某一进程修改了共享对象之后，所有的进程拿到额共享对象都得到了更新。

Manager() 返回的管理器支持的范例有：

list、 dict、 Namespace 、 Lock、 RLock、Semaphore、 BoundedSemaphore、 Condition 、Event、 Barrier、 Queue、 Value 和 Array。

复制代码

示例代码如下，主进程界说了一个字典和一个列表，在子进程中，可以添加和修改字典的内容，在列表中插入新的数据，实现进程间的数据共享，即可以共同修改同一份数据。

from multiprocessing import Process, Manager
def fun1(dic,lis,index):
dic[index] = 'a'
dic['2'] = 'b'
lis.append(index)
if __name__ == '__main__':
with Manager() as manager:
_# 注意字典的声明方式，不能直接通过{}来定义_
dic = manager.dict()
_# [0,1,2,3,4]_
l = manager.list(range(5))
process_list = []
for i in range(10):
p = Process(target=fun1, args=(dic,l,i))
p.start()
process_list.append(p)
for res in process_list:
res.join()
print(dic)
print(l)

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

一文速通Python并行计算：10 Python多进程编程-进程之间的数据共享-基于共 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云