python3學習筆記之多進程分布式小例子

2020-01-04 15:50:49

字體：大中小

來源：轉載

供稿：網友

最近一直跟著廖大在學Python，關于分布式進程的小例子挺有趣的，這里做個記錄。

分布式進程

Python的multiprocessing模塊不但支持多進程，其中managers子模塊還支持把多進程分布到多臺機器上。一個服務進程可以作為調度者，將任務分布到其他多個進程中，依靠網絡通信。由于managers模塊封裝很好，不必了解網絡通信的細節，就可以很容易地編寫分布式多進程程序。

master服務端原理：通過managers模塊把Queue通過網絡暴露出去，其他機器的進程就可以訪問Queue了
服務進程負責啟動Queue，把Queue注冊到網絡上，然后往Queue里面寫入任務，代碼如下:

#task_master.py#coding=utf-8#多進程分布式例子#服務器端from multiprocessing.managers import BaseManagerfrom multiprocessing import freeze_support #server啟動報錯，提示需要引用此包import random,time,queue#發送任務的隊列task_queue = queue.Queue()#接收結果的隊列result_queue = queue.Queue()#從BaseManager繼承的QueueManagerclass QueueManager(BaseManager):  pass#win7 64 貌似不支持callable下調用匿名函數lambda，這里封裝一下def return_task_queue():  global task_queue  return task_queuedef return_result_queue():  global result_queue  return result_queuedef test():  #把兩個Queue注冊到網絡上，callable參數關聯了Queue對象  #QueueManager.register('get_task_queue',callable=lambda:task_queue)  #QueueManager.register('get_result_queue',callable=lambda:result_queue)  QueueManager.register('get_task_queue',callable=return_task_queue)  QueueManager.register('get_result_queue',callable=return_result_queue)  #綁定端口5000，設置驗證碼‘abc'  manager = QueueManager(address=('127.0.0.1',5000),authkey=b'abc')#這里必須加上本地默認ip地址127.0.0.1  #啟動Queue  manager.start()  #server = manager.get_server()  #server.serve_forever()  print('start server master')  #獲得通過網絡訪問的Queue對象  task = manager.get_task_queue()  result = manager.get_result_queue()  #放幾個任務進去  for i in range(10):    n = random.randint(0,10000)    print('put task %d...' % n)    task.put(n)  #從result隊列讀取結果  print('try get results...')  for i in range(10):    r = result.get(timeout=10)    print('result:%s' % r)  #關閉  manager.shutdown()  print('master exit')if __name__ == '__main__':  freeze_support()  test()

運行截圖如下：

python3,多進程,分布式

在分布式多進程環境下，添加任務到Queue不可以直接對原始的task_queue進行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue接口添加。

任務進程,代碼如下：

#task_worker.py#coding=utf-8#多進程分布式例子#非服務端：workerimport time,sys,queuefrom multiprocessing.managers import BaseManager#創建類似的QueueManagerclass QueueManager(BaseManager):  pass#由于這個QueueManager只從網絡上獲取Queue，所以注冊時只提供名字即可QueueManager.register('get_task_queue')QueueManager.register('get_result_queue')#連接到服務器，也就是運行task_master.py的機器server_addr = '127.0.0.1'print('connect to server %s...'% server_addr)#端口和驗證碼注意要保持完全一致m = QueueManager(address=(server_addr,5000),authkey=b'abc')#從網絡連接m.connect()#獲取Queue的對象task = m.get_task_queue()result = m.get_result_queue()#從task隊列獲取任務，并把結果寫入result隊列for i in range(10):  try:    n = task.get(timeout=1)    print('run task %d * %d...'% (n,n))    r = '%d * %d = %d' % (n,n,n*n)    time.sleep(1)    result.put(r)  except queue.Empty:    print('task queue is empty')#處理結果print('worker exit')

運行截圖如下：

python3,多進程,分布式