python多进程通信、共享变量
python多进程通信、共享变量
Python的多进程编程可以充分利用多核CPU的优势，提高程序的运行效率。下面是一个关于Python多进程的介绍和示例。
什么是多进程？
进程是计算机中正在运行的程序的实例。多进程指的是同时运行多个进程。
为什么要用多进程？
由于计算机的CPU是单核的，所以一次只能执行一个任务。 但是现代计算机通常都有多个核心，如果只有一个进程在运行，那么其他核心就处于闲置状态。 多进程编程可以同时利用多个核心，提高程序的运行效率。
Python中的多进程
Python 提供了multiprocessing模块来实现多进程编程。
该模块与threading模块的API相似，但是使用起来更加方便。下面是一个使用multiprocessing模块的示例代码：
import multiprocessing
def worker(num):
"""进程 worker 函数"""
print('Worker %s starting...' % num)
return
if __name__ == '__main__':
# 创建进程
p1 = multiprocessing.Process(target=worker, args=(1,))
p2 = multiprocessing.Process(target=worker, args=(2,))
# 启动进程
p1.start()
p2.start()
# 等待进程结束
p1.join()
p2.join()
print('All processes finished.')Worker 2 starting...
Worker 1 starting...
All processes finished.上面的代码中，我们创建了两个进程并启动它们执行worker函数。使用join()方法等待进程结束，并打印出输出。
多进程中的进程通信
多个进程之间如何通信呢？
Python提供了多种方式，包括管道(Pipe)、队列(Queue)、共享内存(Value和Array)等。
下面是使用队列来在多进程中传递数据的示例代码：
import multiprocessing
from multiprocessing import Lock
def producer(idx, var, queue, lock):
"""生产者进程"""
for i in range(1000):
lock.acquire()
var.value += 1
print("%d 生产者生产了：%d" % (idx, var.value))
lock.release()
queue.put(var.value)
def consumer(idx, queue, lock):
"""消费者进程"""
while True:
if not queue.empty():
item = queue.get()
lock.acquire()
print("%d 消费者消费了：%d" % (idx, item))
lock.release()
else:
pass
if __name__ == '__main__':
lock = Lock()
var = multiprocessing.Value('i', 0)
# 创建队列
queue = multiprocessing.Queue()
# 创建进程
p1 = multiprocessing.Process(target=producer, args=(1, var, queue, lock))
p11 = multiprocessing.Process(target=producer, args=(2, var, queue, lock))
p2 = multiprocessing.Process(target=consumer, args=(1, queue, lock))
p22 = multiprocessing.Process(target=consumer, args=(2, queue, lock))
# 启动进程
p1.start()
p11.start()
p2.start()
p22.start()
# 等待进程结束
p1.join()
p11.join()
p2.join()
p22.join()
print('All processes finished.')上面的代码中，我们创建了一个队列，并将它传递给多个进程。
生产者进程向队列中不断地生产数据，消费者进程则不断地从队列中消费数据。
这样就实现了多个进程之间的通信。
加锁是同一时间只有一个打印能输出，防止日志重叠
回调函数
使用multiprocessing模块创建进程。然后，我们使用Value和Array来创建共享变量，这些变量可以被多个进程访问和修改。
最后，我们还会使用callback函数，这是一个可以在进程完成后执行的函数。
import multiprocessing as mp
# 共享变量
var = mp.Value('i', 0)
arr = mp.Array('i', [0, 1, 2])
# 回调函数
def callback(result):
print("Result:", result)
# 子进程函数
def sub_process(var, arr, callback):
# 修改共享变量
var.value += 1
arr[0] += 10
# 调用回调函数
callback(var.value + arr[0])
if __name__ == '__main__':
# 创建子进程
p1 = mp.Process(target=sub_process, args=(var, arr, callback))
# 启动子进程
p1.start()
# 等待子进程结束
p1.join()
# 打印共享变量的值
print("Var:", var.value)
print("Array:", arr[:])Result: 11
Var: 1
Array: [10, 1, 2]在上面的代码中，我们使用Value和Array分别创建了一个整型变量var和一个整型数组arr。
在子进程函数sub_process中，我们修改了这两个共享变量，并调用了回调函数callback。
在主进程中，我们启动了子进程，并等待子进程完成。完成后，我们打印了修改后的共享变量的值。
进程池共享变量
import multiprocessing as mp
# 定义任务函数
def task(num, shared_var, lock):
for i in range(num):
# 在修改共享变量之前先获取锁
lock.acquire()
shared_var.value -= 1
lock.release()
# 在修改完成之后释放锁
# d = 10 / shared_var.value
return {"shared_var.value": shared_var.value}
def cb(result):
print(f'result: {result}')
def error_cb(result):
print(f'error: {result}')
def main():
lock = mp.Manager().Lock()
shared_var = mp.Manager().Value('i', 10) # 共享变量
# 创建一个进程池
with mp.Pool(processes=4) as pool:
# 提交4个任务，每个任务都会执行10次减1操作
results = [pool.apply_async(task, args=(10, shared_var, lock), callback=cb, error_callback=error_cb) for _ in range(4)]
# 等待所有任务完成
[result.get() for result in results]
# 进程池任务完成后，输出共享变量的值
print("Shared variable value: ", shared_var.value)
if __name__ == '__main__':
main()result: {'shared_var.value': 0}
result: {'shared_var.value': -10}
result: {'shared_var.value': -25} # 中间的结果可能不是 -10, -20
result: {'shared_var.value': -30}
Shared variable value: -30以上就是Python多进程编程的基本内容。如果想了解更多细节，请阅读Python官方文档中有关 multiprocessing 模块的内容。