關於博主

努力與運動兼備～~~有任何問題可以加我好友或者關注微信公眾號，歡迎交流，我們一起進步！

                                      微信公眾號:  啃餅思錄

                                    QQ: 2810706745（啃餅小白）

寫在前面

小驚大怪

你是不是在用Python3或者在windows系統上程式設計？最重要的是你對程序和執行緒不是很清楚？那麼恭喜你，在python分散式程序中，會有坑等著你去挖。。。(hahahaha，此處允許我嚇唬一下你)開玩笑的啦，不過，如果你知道序列中不支援匿名函式，那這個坑就和你say byebye了。好了話不多數，直接進入正題。

分散式程序

正如大家所知道的Process比Thread更穩定，而且Process可以分佈到多臺機器上，而Thread最多隻能分佈到同一臺機器的多個CPU上。Python的multiprocessing模組不但支援多程序，其中managers子模組還支援把多程序分佈到多臺機器上。一個服務程序可以作為排程者，將任務分佈到其他多個程序中，依靠網路通訊。由於managers模組封裝很好，不必瞭解網路通訊的細節，就可以很容易地編寫分散式多程序程式。

程式碼記錄

舉個例子

如果我們已經有一個通過Queue通訊的多程序程式在同一臺機器上執行，現在，由於處理任務的程序任務繁重，希望把傳送任務的程序和處理任務的程序分佈到兩臺機器上，這應該怎麼用分散式程序來實現呢？你已經知道了原有的Queue可以繼續使用，而且通過managers模組把Queue通過網路暴露出去，就可以讓其他機器的程序來訪問Queue了。好，那我們就這麼幹！

寫個task_master.py

我們先看服務程序。服務程序負責啟動Queue，把Queue註冊到網路上，然後往Queue裡面寫入任務。

#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_master.py

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

# 傳送任務的佇列:
task_queue = queue.Queue()
# 接收結果的佇列:
 

result_queue = queue.Queue()

# 從BaseManager繼承的QueueManager:
class QueueManager(BaseManager):
    pass

# 把兩個Queue都註冊到網路上, callable引數關聯了Queue物件:
QueueManager.register('get_task_queue', callable=lambda: 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 results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)
# 關閉:
manager.shutdown()
print('master exit.')

請注意，當我們在一臺機器上寫多程序程式時，建立的Queue可以直接拿來用，但是，在分散式多程序環境下，新增任務到Queue不可以直接對原始的task_queue進行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue介面新增。然後，在另一臺機器上啟動任務程序（本機上啟動也可以）

寫個task_worker.py

#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_worker.py

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('Connect to server %s...' % server_addr)
# 埠和驗證碼注意保持與task_master.py設定的完全一致:
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.')

任務程序要通過網路連線到服務程序，所以要指定服務程序的IP。

執行結果

現在，可以試試分散式程序的工作效果了。先啟動task_master.py服務程序：

Traceback (most recent call last):
  File "F:/Python/untitled/xianchengjincheng/master.py", line 25, in <module>
    manager.start()
  File "F:\Python\pystall\lib\multiprocessing\managers.py", line 513, in start
    self._process.start()
  File "F:\Python\pystall\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "F:\Python\pystall\lib\multiprocessing\context.py", line 322, in _Popen
    return Popen(process_obj)
  File "F:\Python\pystall\lib\multiprocessing\popen_spawn_win32.py", line 65, in __init__
    reduction.dump(process_obj, to_child)
  File "F:\Python\pystall\lib\multiprocessing\reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <function <lambda> at 0x00000202D1921E18>: attribute lookup <lambda> on __main__ failed

task_master.py程序傳送完任務後，開始等待result佇列的結果。現在啟動task_worker.py程序：

Connect to server 127.0.0.1...
Traceback (most recent call last):
  File "F:/Python/untitled/xianchengjincheng/work.py", line 24, in <module>
    m.connect()
  File "F:\Python\pystall\lib\multiprocessing\managers.py", line 489, in connect
    conn = Client(self._address, authkey=self._authkey)
  File "F:\Python\pystall\lib\multiprocessing\connection.py", line 487, in Client
    c = SocketClient(address)
  File "F:\Python\pystall\lib\multiprocessing\connection.py", line 614, in SocketClient
    s.connect(address)
ConnectionRefusedError: [WinError 10061] 由於目標計算機積極拒絕，無法連線。

看到沒，結果都出錯了，我們好好分析一下到底哪出錯了。。。

錯誤分析

在task_master.py的報錯提示中，我們知道它說lambda錯誤，這是因為序列化不支援匿名函式，所以我們得修改程式碼，重新對queue用QueueManager進行封裝放到網路中。

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

其中task_queue和result_queue是兩個佇列，分別存放任務和結果。它們用來進行程序間通訊，交換物件。 因為是分散式的環境，放入queue中的資料需要等待Workers機器運算處理後再進行讀取，這樣就需要對queue用QueueManager進行封裝放到網路中，這是通過上面的2行程式碼來實現的。我們給return_task_queue的網路呼叫介面取了一個名get_task_queue,而return_result_queue的名字是get_result_queue，方便區分對哪個queue進行操作。task.put(n)即是對task_queue進行寫入資料，相當於分配任務。而result.get()即是等待workers機器處理後返回的結果。

值得注意 在windows系統中你必須要寫IP地址，而其他作業系統比如linux作業系統則就不要了。

 # windows需要寫ip地址
 manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')

修改後的程式碼

在task_master.py中修改如下：

#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_master.py

# task_master.py

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

task_queue =  queue.Queue()  # 傳送任務的佇列:
result_queue = queue.Queue() # 接收結果的佇列:
class QueueManager(BaseManager):  # 從BaseManager繼承的QueueManager:
    pass
# windows下執行
def return_task_queue():
    global task_queue
    return task_queue  # 返回傳送任務佇列
def return_result_queue ():
    global result_queue
    return result_queue # 返回接收結果佇列

def 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=('', 5000), authkey=b'abc')
    # windows需要寫ip地址
    manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
    manager.start()  # 啟動Queue:
    # 獲得通過網路訪問的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):
        # 這裡加了異常捕獲
        try:
            r = result.get(timeout=5)
            print('Result: %s' % r)
        except queue.Empty:
             print('result queue is empty.')
    # 關閉:
    manager.shutdown()
    print('master exit.')
if __name__=='__main__':
    freeze_support()
    print('start!')
    test()

在task_worker.py中修改如下：

#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_worker.py


# task_worker.py

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('Connect to server %s...' % server_addr)
# 埠和驗證碼注意保持與task_master.py設定的完全一致:
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.')

先執行task_master.py,然後再執行task_worker.py (1)task_master.py執行結果如下

start!
Put task 7872...
Put task 6931...
Put task 1395...
Put task 8477...
Put task 8300...
Put task 1597...
Put task 8738...
Put task 8627...
Put task 1884...
Put task 2561...
Try get results...
Result: 7872 * 7872 = 61968384
Result: 6931 * 6931 = 48038761
Result: 1395 * 1395 = 1946025
Result: 8477 * 8477 = 71859529
Result: 8300 * 8300 = 68890000
Result: 1597 * 1597 = 2550409
Result: 8738 * 8738 = 76352644
Result: 8627 * 8627 = 74425129
Result: 1884 * 1884 = 3549456
Result: 2561 * 2561 = 6558721
master exit.

(2)task_worker.py執行結果如下

Connect to server 127.0.0.1...
run task 8640 * 8640...
run task 7418 * 7418...
run task 9303 * 9303...
run task 568 * 568...
run task 1633 * 1633...
run task 3583 * 3583...
run task 3293 * 3293...
run task 8975 * 8975...
run task 8189 * 8189...
run task 731 * 731...
worker exit.

知識補充

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

Queue物件儲存在哪？注意到task_worker.py中根本沒有建立Queue的程式碼，所以，Queue物件儲存在task_master.py程序中：

而Queue之所以能通過網路訪問，就是通過QueueManager實現的。由於QueueManager管理的不止一個Queue，所以，要給每個Queue的網路呼叫介面起個名字，比如get_task_queue。task_worker這裡的QueueManager註冊的名字必須和task_manager中的一樣。對比上面的例子，可以看出Queue物件從另一個程序通過網路傳遞了過來。只不過這裡的傳遞和網路通訊由QueueManager完成。

authkey有什麼用？這是為了保證兩臺機器正常通訊，不被其他機器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連線不上。

今天的分享就到這裡了，如果你有任何不懂的問題，可以發信息或者留言嘍。