分散式程序

Python的multiprocessing模組不但支援多程序，其中的managers子模組還支援吧多程序分佈到多臺機器上。

Demo

執行示例

服務端

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# 分佈程序-服務端

import random, time, queue
from multiprocessing.managers import BaseManager

# 傳送任務的佇列
task_queue = queue.Queue()

# 接收任務的佇列
result_queue = queue.Queue()

# 從 BaseManager繼承的 QueueManager
 

class QueueManager(BaseManager):
    pass

def get_task_queue():
   global task_queue
   return task_queue

# 把兩個Queue都註冊到網路上,callable引數關聯了 Queue 物件
QueueManager.register("get_task_queue", callable = get_task_queue)
QueueManager.register("get_result_queue", callable = lambda:result_queue)

# 繫結埠5000，設定驗證碼為"abc"
 

manager = QueueManager(address=("", 5000), authkey=b"abc")

# 啟動 Queue
manager.start()

# 獲取通過網路訪問的 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 result"
 
)
for i in range(10):
    r = result.get(timeout = 10)
    print("結果 %s" % r)

# 關閉
manager.shutdown()
print("manage exit")

客戶端

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# 分佈程序-客戶端
import time, sys, queue
from multiprocessing.managers import BaseManager

# 建立類似的 QueueManager
class 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("連線伺服器 %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" %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")

Window無法正常執行，例如

D:\PythonProject>Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\spawn.py", line 99, in spawn_main
    new_handle = reduction.steal_handle(parent_pid, pipe_handle)
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\reduction.py", line 82, in steal_handle
    _winapi.PROCESS_DUP_HANDLE, False, source_pid)
OSError: [WinError 87] 引數錯誤。


python task_worker.py
連線伺服器 127.0.0.1
Traceback (most recent call last):
  File "task_worker.py", line 23, in <module>
    m.connect()
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\managers.py", line 489, in connect
    conn = Client(self._address, authkey=self._authkey)
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 487, in Client
    c = SocketClient(address)
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 614, in SocketClient
    s.connect(address)
ConnectionRefusedError: [WinError 10061] 由於目標計算機積極拒絕，無法連線。

這個簡單的Master/Worker模型有什麼用？其實這就是一個簡單但真正的分散式計算，把程式碼稍加改造，啟動多個worker，就可以把任務分佈到幾臺甚至幾十臺機器上，比如把計算n*n的程式碼換成傳送郵件，就實現了郵件佇列的非同步傳送。

看來這個demo暫時無法看效果了，可惜