利用Python multiprocessing实现分布式任务架构

Finn

发布时间：2026-10-13| 2人浏览过

原创

本文介绍如何利用Python内置的multiprocessing模块中的BaseManager类，快速构建基于生产者-消费者模式的分布式任务系统。通过中央节点协调，无需引入外部队列中间件，即可实现多主机间的高效通信与任务分发。

深入研读Python官方文档后发现，内置的multiprocessing模块提供了丰富的预制接口，能够便捷地实现多台主机间的通信，从而轻松搭建典型的生产者-消费者模式的分布式任务架构。

以往在Python中实现生产者-消费者模式时，通常需借助额外的队列系统（如RabbitMQ等）。此外，还需自行设计任务对象的序列化方案以存入队列。若缺乏队列支持，部分开发者可能不得不从Socket服务器底层开发做起，直接处理TCP/IP网络通信。

实际上，multiprocessing.managers模块中的BaseManager类为开发者提供了这样一个快速实现的接口。

假设的场景包含1个生产者（producer.py）和8个消费者（worker.py），并设有一个中央协调节点（server.py），具体实现如下：

server.py

from multiprocessing.managers import BaseManager
import Queue

queue = Queue.Queue() #初始化一个Q，用于消息传递
class QueueManager(BaseManager):
  pass

QueueManager.register('get_queue', callable=lambda:queue) # 在系统中发布get_queue这个业务

if __name__ == '__main__':
  m = QueueManager(address=('10.239.85.193', 50000),authkey='abr' )
 # 监听所有10.239.85.193的50000口
  s = m.get_server()
  s.serve_forever()

worker.py

from multiprocessing.managers import BaseManager
from multiprocessing import Pool


class QueueManager(BaseManager):
 pass

QueueManager.register('get_queue') 

def feb(i): #经典的'山羊增殖'
  if i < 2: return 1
  if i < 5 : return feb(i-1) + feb(i-2)
  return feb(i-1) + feb(i-2) - feb(i-5)

def worker(i): 
  m = QueueManager(address=('10.239.85.193', 50000), authkey='abr')
#连接server
  m.connect()
  while True:
    queue = m.get_queue()
# 获取Q
   c = queue.get()
 print feb(c)

if __name__ == '__main__':

  p = Pool(8) # 分进程启动8个worker
  p.map(worker, range(8))
producer.py

from multiprocessing.managers import BaseManager


class QueueManager(BaseManager):
  pass
QueueManager.register('get_queue')


if __name__ == '__main__':
 m = QueueManager(address=('10.239.85.193', 50000), authkey='abr')
 m.connect()
 i = 0
 while True:
   queue = m.get_queue()
   queue.put(48)

   i+=1

系统会将Queue()对象中的数据直接封装，并通过TCP 50000端口在主机间传递。需注意，由于authkey机制的限制，各节点必须保持Python版本一致。

以上内容为本文全部介绍，希望对读者的学习有所助益，也欢迎大家继续支持本站。