Python多进程与分布式数据共享实战：从基础到Manager应用
本文详解Python多进程基础用法，包括进程创建、PID区分及join同步机制。重点介绍利用Manager实现分布式环境下的数据共享，通过BaseManager搭建服务端与客户端连接，演示多子进程修改共享列表的完整流程，解决跨进程数据同步问题。
本文通过具体实例，深入解析Python多进程的基础概念及其在分布式环境下的数据共享机制，旨在为开发者提供清晰的参考与实操指导。
Python多进程基础入门
Python多进程基础入门
官方文档参考：https://docs.python.org/3/library/multiprocessing.html
1、基础示例
# coding: utf-8
from multiprocessing import Process
# 定义函数
def addUser():
print("addUser")
if __name__ == "__main__":
p1 = Process(target=addUser)
p1.start()
引入多进程模块multiprocessing；
实例化进程对象p1 = Process(target=函数名);
启动进程执行任务p1.start()。
2、通过进程ID区分父子进程
# coding: utf-8
from multiprocessing import Process
import os
# 定义一个list
myList = ["a","b"];
# 定义函数
def addUser(list):
print(os.getpid()) # 进程ID
print(os.getppid()) # 父进程ID
if __name__ == "__main__":
p1 = Process(target=addUser,args=(myList,))
p1.start()
# 这里是主进程
print("父进程ID:"+str(os.getpid())) # 进程ID
父进程ID:27084 子进程ID:27085 父进程ID:27084
3、主进程等待子进程执行完毕
# 定义一个list
myList = ["a","b"];
# 定义函数
def addUser(list):
list.append("c")
print(list)
if __name__ == "__main__":
p1 = Process(target=addUser,args=(myList,))
p1.start()
print(myList)
['a', 'b'] ['a', 'b', 'c']
主线程中的print(myList)操作先于子进程中的print(list)操作执行。
若需确保主进程等待子进程完成，只需在主进程代码中增加一行：
p1.join() # 等待子进程执行完毕
print(myList)
['a', 'b', 'c'] ['a', 'b']
此时，代码的执行顺序将发生相应变化。
分布式进程数据共享实战
分布式进程数据共享方案
利用Manager机制来实现多进程间的数据共享。
Manager会启动一个独立的服务进程，所有其他工作进程统一通过该Server进程访问共享数据，从而实现多进程间的数据通信。
需要注意的是，一旦主进程结束，该Server服务进程也将随之终止。
1、直接展示核心代码实现
# coding: utf-8
from multiprocessing import Process, Manager
# 定义函数
def addUser(list):
list.append("c") # 给list添加了一个元素
print(list)
if __name__ == "__main__":
mgr = Manager()
my_list = mgr.list(["a","b"]) # 通过Manager对象创建list
p1 = Process(target=addUser,args=(my_list,))
p1.start()
p1.join()
print(my_list)
注意代码中两次打印list列表，其中子进程(addUser())内部对list数据进行了修改。观察这两次打印的结果：
['a', 'b', 'c'] ['a', 'b', 'c']
上述结果证实，通过Manager对象创建的list数据已成功在进程间实现了通信与共享。
2、分布式数据共享完整流程 https://docs.python.org/3/library/multiprocessing.html
2.1、构建Server端
datamgr.py文件内容如下：
# coding: utf-8
from multiprocessing.managers import BaseManager
if __name__ == "__main__":
mgr = BaseManager(address=('127.0.0.1', 50000), authkey=b'password')
mgr.register("getUser", callable=lambda :["a","b"])
# server永不关闭
server = mgr.get_server()
server.serve_forever()
该脚本作为数据提供者运行。
2.2、在test.py中（可能运行于另一台服务器）连接该Server
# coding: utf-8
from multiprocessing import Process, Manager
from multiprocessing.managers import BaseManager
if __name__ == "__main__":
mgr = BaseManager(address=('127.0.0.1', 50000), authkey=b'password')
mgr.register("getUser")
mgr.connect() # 连接远程的server
my_list = mgr.getUser() # 从server获取数据
print(my_list) # ['a', 'b']
连接建立成功，并成功获取到数据。
3、创建2个子进程修改list数据，验证共享效果
# 定义函数
def addUser(list):
list.append("c")
def addUser2(list):
list.append("d")
if __name__ == "__main__":
mgr = BaseManager(address=('127.0.0.1', 50000), authkey=b'password')
mgr.register("getUser")
mgr.connect() # 连接远程的server
my_list = mgr.getUser() # 从server获取数据
# 创建2个子进程
p1 = Process(target=addUser,args=(my_list,))
p1.start()
p2 = Process(target=addUser2, args=(my_list,))
p2.start()
# 等待2个子进程执行完毕
p1.join()
p2.join()
# 在主进程打印最终的list
print(my_list) # ['a', 'b', 'c', 'd']
结果显示，在子进程中向list添加数据的功能正常生效。
更多关于Python相关内容感兴趣的读者可查看本站专题：《Python进程与线程操作技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》、《Python+MySQL数据库程序设计入门教程》及《Python常见数据库操作技巧汇总》
希望本文所述对大家Python程序设计有所帮助。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
