Python如何快速實現分布式任務

更新時間：2017年07月06日 09:41:42 作者：開源小站

這篇文章主要介紹了Python如何快速實現分布式任務,小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

深入讀了讀python的官方文檔，發(fā)覺Python自帶的multiprocessing模塊有很多預制的接口可以方便的實現多個主機之間的通訊，進而實現典型的生產者-消費者模式的分布式任務架構。

之前，為了在Python中實現生產者-消費者模式，往往就會選擇一個額外的隊列系統(tǒng)，比如rabbitMQ之類。此外，你有可能還要設計一套任務對象的序列化方式以便塞入隊列。如果沒有隊列的支持，那不排除有些同學不得不從socket服務器做起，直接跟TCP/IP打起交道來。

其實multiprocessing.managers中有個BaseManager就為開發(fā)者提供了這樣一個快速接口。

我們假定的場景是1個生產者（producer.py）+8個消費者(worker.py)的系統(tǒng)，還有一個中央節(jié)點負責協(xié)調（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) # 在系統(tǒng)中發(fā)布get_queue這個業(yè)務

if __name__ == '__main__':
  m = QueueManager(address=('10.239.85.193', 50000),authkey='abr' )
 # 監(jiān)聽所有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

系統(tǒng)會直接將Queue() 對象中的數據直接封裝后通過TCP 50000端口在主機之間傳遞。不過需要注意的是，由于authkey的緣故，各個節(jié)點要求python的版本一致。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: