日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間：8:30-17:00
你可能遇到了下面的問(wèn)題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
Python分布式進(jìn)程中你會(huì)遇到的坑

python分布式進(jìn)程中你會(huì)遇到的坑

作者：編程思錄 2019-05-28 08:56:40

開發(fā)

后端

分布式 你是不是在用Python3或者在windows系統(tǒng)上編程?最重要的是你對(duì)進(jìn)程和線程不是很清楚?

 [[266539]]

小驚大怪

你是不是在用Python3或者在windows系統(tǒng)上編程?最重要的是你對(duì)進(jìn)程和線程不是很清楚?那么恭喜你，在python分布式進(jìn)程中，會(huì)有坑等著你去挖。。。(hahahaha，此處允許我嚇唬一下你)開玩笑的啦，不過(guò)，如果你知道序列中不支持匿名函數(shù)，那這個(gè)坑就和你say byebye了。好了話不多數(shù)，直接進(jìn)入正題。

分布式進(jìn)程

正如大家所知道的Process比thread更穩(wěn)定，而且Process可以分布到多臺(tái)機(jī)器上，而Thread最多只能分布到同一臺(tái)機(jī)器的多個(gè)CPU上。Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺(tái)機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個(gè)進(jìn)程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細(xì)節(jié)，就可以很容易地編寫分布式多進(jìn)程程序。

代碼記錄

舉個(gè)例子

如果我們已經(jīng)有一個(gè)通過(guò)Queue通信的多進(jìn)程程序在同一臺(tái)機(jī)器上運(yùn)行，現(xiàn)在，由于處理任務(wù)的進(jìn)程任務(wù)繁重，希望把發(fā)送任務(wù)的進(jìn)程和處理任務(wù)的進(jìn)程分布到兩臺(tái)機(jī)器上，這應(yīng)該怎么用分布式進(jìn)程來(lái)實(shí)現(xiàn)呢?你已經(jīng)知道了原有的Queue可以繼續(xù)使用，而且通過(guò)managers模塊把Queue通過(guò)網(wǎng)絡(luò)暴露出去，就可以讓其他機(jī)器的進(jìn)程來(lái)訪問(wèn)Queue了。好，那我們就這么干!

寫個(gè)task_master.py

我們先看服務(wù)進(jìn)程。服務(wù)進(jìn)程負(fù)責(zé)啟動(dòng)Queue，把Queue注冊(cè)到網(wǎng)絡(luò)上，然后往Queue里面寫入任務(wù)。

  
 
 
 
    
   
  
  
  
  1. #！/user/bin/pytthon 
    2.    
  
  
  
  2. # -*- coding:utf-8 -*- 
    3.    
  
  
  
  3. # @Time: 2018/3/3 16:46 
    4.    
  
  
  
  4. # @Author: lichexo 
    5.    
  
  
  
  5. # @File: task_master.py 
    6.    
  
  
  
  6. import random, time, queue 
    7.    
  
  
  
  7. from multiprocessing.managers import BaseManager 
    8.    
  
  
  
  8. # 發(fā)送任務(wù)的隊(duì)列: 
    9.    
  
  
  
  9. task_queue = queue.Queue() 
    10.    
  
  
  
  10. # 接收結(jié)果的隊(duì)列: 
    11.    
  
  
  
  11. result_queue = queue.Queue() 
    12.    
  
  
  
  12. # 從BaseManager繼承的QueueManager: 
    13.    
  
  
  
  13. class QueueManager(BaseManager): 
    14.    
  
  
  
  14.  pass 
    15.    
  
  
  
  15. # 把兩個(gè)Queue都注冊(cè)到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象: 
    16.    
  
  
  
  16. QueueManager.register('get_task_queue', callable=lambda: task_queue) 
    17.    
  
  
  
  17. QueueManager.register('get_result_queue', callable=lambda: result_queue) 
    18.    
  
  
  
  18. # 綁定端口5000, 設(shè)置驗(yàn)證碼'abc': 
    19.    
  
  
  
  19. manager = QueueManager(address=('', 5000), authkey=b'abc') 
    20.    
  
  
  
  20. # 啟動(dòng)Queue: 
    21.    
  
  
  
  21. manager.start() 
    22.    
  
  
  
  22. # 獲得通過(guò)網(wǎng)絡(luò)訪問(wèn)的Queue對(duì)象: 
    23.    
  
  
  
  23. task = manager.get_task_queue() 
    24.    
  
  
  
  24. result = manager.get_result_queue() 
    25.    
  
  
  
  25. # 放幾個(gè)任務(wù)進(jìn)去: 
    26.    
  
  
  
  26. for i in range(10): 
    27.    
  
  
  
  27.  n = random.randint(0, 10000) 
    28.    
  
  
  
  28.  print('Put task %d...' % n) 
    29.    
  
  
  
  29.  task.put(n) 
    30.    
  
  
  
  30. # 從result隊(duì)列讀取結(jié)果: 
    31.    
  
  
  
  31. print('Try get results...') 
    32.    
  
  
  
  32. for i in range(10): 
    33.    
  
  
  
  33.  r = result.get(timeout=10) 
    34.    
  
  
  
  34.  print('Result: %s' % r) 
    35.    
  
  
  
  35. # 關(guān)閉: 
    36.    
  
  
  
  36. manager.shutdown() 
    37.    
  
  
  
  37. print('master exit.') 
    38.

請(qǐng)注意，當(dāng)我們?cè)谝慌_(tái)機(jī)器上寫多進(jìn)程程序時(shí)，創(chuàng)建的Queue可以直接拿來(lái)用，但是，在分布式多進(jìn)程環(huán)境下，添加任務(wù)到Queue不可以直接對(duì)原始的task_queue進(jìn)行操作，那樣就繞過(guò)了QueueManager的封裝，必須通過(guò)manager.get_task_queue()獲得的Queue接口添加。然后，在另一臺(tái)機(jī)器上啟動(dòng)任務(wù)進(jìn)程(本機(jī)上啟動(dòng)也可以)

寫個(gè)task_worker.py

  
 
 
 
    
   
  
  
  
  1. #！/user/bin/pytthon 
    2.    
  
  
  
  2. # -*- coding:utf-8 -*- 
    3.    
  
  
  
  3. # @Time: 2018/3/3 16:46 
    4.    
  
  
  
  4. # @Author: lichexo 
    5.    
  
  
  
  5. # @File: task_worker.py 
    6.    
  
  
  
  6. import time, sys, queue 
    7.    
  
  
  
  7. from multiprocessing.managers import BaseManager 
    8.    
  
  
  
  8. # 創(chuàng)建類似的QueueManager: 
    9.    
  
  
  
  9. class QueueManager(BaseManager): 
    10.    
  
  
  
  10.  pass 
    11.    
  
  
  
  11. # 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊(cè)時(shí)只提供名字: 
    12.    
  
  
  
  12. QueueManager.register('get_task_queue') 
    13.    
  
  
  
  13. QueueManager.register('get_result_queue') 
    14.    
  
  
  
  14. # 連接到服務(wù)器，也就是運(yùn)行task_master.py的機(jī)器: 
    15.    
  
  
  
  15. server_addr = '127.0.0.1' 
    16.    
  
  
  
  16. print('Connect to server %s...' % server_addr) 
    17.    
  
  
  
  17. # 端口和驗(yàn)證碼注意保持與task_master.py設(shè)置的完全一致: 
    18.    
  
  
  
  18. m = QueueManager(address=(server_addr, 5000), authkey=b'abc') 
    19.    
  
  
  
  19. # 從網(wǎng)絡(luò)連接: 
    20.    
  
  
  
  20. m.connect() 
    21.    
  
  
  
  21. # 獲取Queue的對(duì)象: 
    22.    
  
  
  
  22. task = m.get_task_queue() 
    23.    
  
  
  
  23. result = m.get_result_queue() 
    24.    
  
  
  
  24. # 從task隊(duì)列取任務(wù),并把結(jié)果寫入result隊(duì)列: 
    25.    
  
  
  
  25. for i in range(10): 
    26.    
  
  
  
  26.  try: 
    27.    
  
  
  
  27.  n = task.get(timeout=1) 
    28.    
  
  
  
  28.  print('run task %d * %d...' % (n, n)) 
    29.    
  
  
  
  29.  r = '%d * %d = %d' % (n, n, n*n) 
    30.    
  
  
  
  30.  time.sleep(1) 
    31.    
  
  
  
  31.  result.put(r) 
    32.    
  
  
  
  32.  except Queue.Empty: 
    33.    
  
  
  
  33.  print('task queue is empty.') 
    34.    
  
  
  
  34. # 處理結(jié)束: 
    35.    
  
  
  
  35. print('worker exit.') 
    36.

任務(wù)進(jìn)程要通過(guò)網(wǎng)絡(luò)連接到服務(wù)進(jìn)程，所以要指定服務(wù)進(jìn)程的IP。

運(yùn)行結(jié)果

現(xiàn)在，可以試試分布式進(jìn)程的工作效果了。先啟動(dòng)task_master.py服務(wù)進(jìn)程：

  
 
 
 
    
   
  
  
  
  1. Traceback (most recent call last): 
    2.    
  
  
  
  2.  File "F:/Python/untitled/xianchengjincheng/master.py", line 25, in  
    3.    
  
  
  
  3.  manager.start() 
    4.    
  
  
  
  4.  File "F:Pythonpystalllibmultiprocessingmanagers.py", line 513, in start 
    5.    
  
  
  
  5.  self._process.start() 
    6.    
  
  
  
  6.  File "F:Pythonpystalllibmultiprocessingprocess.py", line 105, in start 
    7.    
  
  
  
  7.  self._popen = self._Popen(self) 
    8.    
  
  
  
  8.  File "F:Pythonpystalllibmultiprocessingcontext.py", line 322, in _Popen 
    9.    
  
  
  
  9.  return Popen(process_obj) 
    10.    
  
  
  
  10.  File "F:Pythonpystalllibmultiprocessingpopen_spawn_win32.py", line 65, in __init__ 
    11.    
  
  
  
  11.  reduction.dump(process_obj, to_child) 
    12.    
  
  
  
  12.  File "F:Pythonpystalllibmultiprocessing 
    13.    
  
  
  
  13. eduction.py", line 60, in dump 
    14.    
  
  
  
  14.  ForkingPickler(file, protocol).dump(obj) 
    15.    
  
  
  
  15. _pickle.PicklingError: Can't pickle  at 0x00000202D1921E18>: attribute lookup  on __main__ failed 
    16.

task_master.py進(jìn)程發(fā)送完任務(wù)后，開始等待result隊(duì)列的結(jié)果?，F(xiàn)在啟動(dòng)task_worker.py進(jìn)程：

  
 
 
 
    
   
  
  
  
  1. Connect to server 127.0.0.1... 
    2.    
  
  
  
  2. Traceback (most recent call last): 
    3.    
  
  
  
  3.  File "F:/Python/untitled/xianchengjincheng/work.py", line 24, in  
    4.    
  
  
  
  4.  m.connect() 
    5.    
  
  
  
  5.  File "F:Pythonpystalllibmultiprocessingmanagers.py", line 489, in connect 
    6.    
  
  
  
  6.  conn = Client(self._address, authkey=self._authkey) 
    7.    
  
  
  
  7.  File "F:Pythonpystalllibmultiprocessingconnection.py", line 487, in Client 
    8.    
  
  
  
  8.  c = SocketClient(address) 
    9.    
  
  
  
  9.  File "F:Pythonpystalllibmultiprocessingconnection.py", line 614, in SocketClient 
    10.    
  
  
  
  10.  s.connect(address) 
    11.    
  
  
  
  11. ConnectionRefusedError: [WinError 10061] 由于目標(biāo)計(jì)算機(jī)積極拒絕，無(wú)法連接。 
    12.

看到?jīng)]，結(jié)果都出錯(cuò)了，我們好好分析一下到底哪出錯(cuò)了。。。

錯(cuò)誤分析

在task_master.py的報(bào)錯(cuò)提示中，我們知道它說(shuō)lambda錯(cuò)誤，這是因?yàn)樾蛄谢恢С帜涿瘮?shù)，所以我們得修改代碼，重新對(duì)queue用QueueManager進(jìn)行封裝放到網(wǎng)絡(luò)中。

  
 
 
 
    
   
  
  
  
  1. # 把兩個(gè)Queue都注冊(cè)到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象 
    2.    
  
  
  
  2. QueueManager.register('get_task_queue',callable=return_task_queue)  
    3.    
  
  
  
  3. QueueManager.register('get_result_queue',callable=return_result_queue) 
    4.

其中task_queue和result_queue是兩個(gè)隊(duì)列，分別存放任務(wù)和結(jié)果。它們用來(lái)進(jìn)行進(jìn)程間通信，交換對(duì)象。

因?yàn)槭欠植际降沫h(huán)境，放入queue中的數(shù)據(jù)需要等待Workers機(jī)器運(yùn)算處理后再進(jìn)行讀取，這樣就需要對(duì)queue用QueueManager進(jìn)行封裝放到網(wǎng)絡(luò)中，這是通過(guò)上面的2行代碼來(lái)實(shí)現(xiàn)的。我們給return_task_queue的網(wǎng)絡(luò)調(diào)用接口取了一個(gè)名get_task_queue,而return_result_queue的名字是get_result_queue，方便區(qū)分對(duì)哪個(gè)queue進(jìn)行操作。task.put(n)即是對(duì)task_queue進(jìn)行寫入數(shù)據(jù)，相當(dāng)于分配任務(wù)。而result.get()即是等待workers機(jī)器處理后返回的結(jié)果。

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

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

修改后的代碼

在task_master.py中修改如下：

  
 
 
 
    
   
  
  
  
  1. #！/user/bin/pytthon 
    2.    
  
  
  
  2. # -*- coding:utf-8 -*- 
    3.    
  
  
  
  3. # @Time: 2018/3/3 16:46 
    4.    
  
  
  
  4. # @Author: lichexo 
    5.    
  
  
  
  5. # @File: task_master.py 
    6.    
  
  
  
  6. # task_master.py 
    7.    
  
  
  
  7. import random,time,queue 
    8.    
  
  
  
  8. from multiprocessing.managers import BaseManager 
    9.    
  
  
  
  9. from multiprocessing import freeze_support 
    10.    
  
  
  
  10. task_queue = queue.Queue() # 發(fā)送任務(wù)的隊(duì)列: 
    11.    
  
  
  
  11. result_queue = queue.Queue() # 接收結(jié)果的隊(duì)列: 
    12.    
  
  
  
  12. class QueueManager(BaseManager): # 從BaseManager繼承的QueueManager: 
    13.    
  
  
  
  13.  pass 
    14.    
  
  
  
  14. # windows下運(yùn)行 
    15.    
  
  
  
  15. def return_task_queue(): 
    16.    
  
  
  
  16.  global task_queue 
    17.    
  
  
  
  17.  return task_queue # 返回發(fā)送任務(wù)隊(duì)列 
    18.    
  
  
  
  18. def return_result_queue (): 
    19.    
  
  
  
  19.  global result_queue 
    20.    
  
  
  
  20.  return result_queue # 返回接收結(jié)果隊(duì)列 
    21.    
  
  
  
  21. def test(): 
    22.    
  
  
  
  22.  # 把兩個(gè)Queue都注冊(cè)到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象,它們用來(lái)進(jìn)行進(jìn)程間通信，交換對(duì)象 
    23.    
  
  
  
  23.  #QueueManager.register('get_task_queue', callable=lambda: task_queue) 
    24.    
  
  
  
  24.  #QueueManager.register('get_result_queue', callable=lambda: result_queue) 
    25.    
  
  
  
  25.  QueueManager.register('get_task_queue', callable=return_task_queue) 
    26.    
  
  
  
  26.  QueueManager.register('get_result_queue', callable=return_result_queue) 
    27.    
  
  
  
  27.  # 綁定端口5000, 設(shè)置驗(yàn)證碼'abc': 
    28.    
  
  
  
  28.  #manager = QueueManager(address=('', 5000), authkey=b'abc') 
    29.    
  
  
  
  29.  # windows需要寫ip地址 
    30.    
  
  
  
  30.  manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc') 
    31.    
  
  
  
  31.  manager.start() # 啟動(dòng)Queue: 
    32.    
  
  
  
  32.  # 獲得通過(guò)網(wǎng)絡(luò)訪問(wèn)的Queue對(duì)象: 
    33.    
  
  
  
  33.  task = manager.get_task_queue() 
    34.    
  
  
  
  34.  result = manager.get_result_queue() 
    35.    
  
  
  
  35.  for i in range(10): # 放幾個(gè)任務(wù)進(jìn)去: 
    36.    
  
  
  
  36.  n = random.randint(0, 10000) 
    37.    
  
  
  
  37.  print('Put task %d...' % n) 
    38.    
  
  
  
  38.  task.put(n) 
    39.    
  
  
  
  39.  # 從result隊(duì)列讀取結(jié)果: 
    40.    
  
  
  
  40.  print('Try get results...') 
    41.    
  
  
  
  41.  for i in range(10): 
    42.    
  
  
  
  42.  # 這里加了異常捕獲 
    43.    
  
  
  
  43.  try: 
    44.    
  
  
  
  44.  r = result.get(timeout=5) 
    45.    
  
  
  
  45.  print('Result: %s' % r) 
    46.    
  
  
  
  46.  except queue.Empty: 
    47.    
  
  
  
  47.  print('result queue is empty.') 
    48.    
  
  
  
  48.  # 關(guān)閉: 
    49.    
  
  
  
  49.  manager.shutdown() 
    50.    
  
  
  
  50.  print('master exit.') 
    51.    
  
  
  
  51. if __name__=='__main__': 
    52.    
  
  
  
  52.  freeze_support() 
    53.    
  
  
  
  53.  print('start!') 
    54.    
  
  
  
  54.  test() 
    55.

在task_worker.py中修改如下：

  
 
 
 
    
   
  
  
  
  1. #！/user/bin/pytthon 
    2.    
  
  
  
  2. # -*- coding:utf-8 -*- 
    3.    
  
  
  
  3. # @Time: 2018/3/3 16:46 
    4.    
  
  
  
  4. # @Author: lichexo 
    5.    
  
  
  
  5. # @File: task_worker.py 
    6.    
  
  
  
  6. # task_worker.py 
    7.    
  
  
  
  7. import time, sys, queue 
    8.    
  
  
  
  8. from multiprocessing.managers import BaseManager 
    9.    
  
  
  
  9. # 創(chuàng)建類似的QueueManager: 
    10.    
  
  
  
  10. class QueueManager(BaseManager): 
    11.    
  
  
  
  11.  pass 
    12.    
  
  
  
  12. # 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊(cè)時(shí)只提供名字: 
    13.    
  
  
  
  13. QueueManager.register('get_task_queue') 
    14.    
  
  
  
  14. QueueManager.register('get_result_queue') 
    15.    
  
  
  
  15. # 連接到服務(wù)器，也就是運(yùn)行task_master.py的機(jī)器: 
    16.    
  
  
  
  16. server_addr = '127.0.0.1' 
    17.    
  
  
  
  17. print('Connect to server %s...' % server_addr) 
    18.    
  
  
  
  18. # 端口和驗(yàn)證碼注意保持與task_master.py設(shè)置的完全一致: 
    19.    
  
  
  
  19. m = QueueManager(address=(server_addr, 5000), authkey=b'abc') 
    20.    
  
  
  
  20. # 從網(wǎng)絡(luò)連接: 
    21.    
  
  
  
  21. m.connect() 
    22.    
  
  
  
  22. # 獲取Queue的對(duì)象: 
    23.    
  
  
  
  23. task = m.get_task_queue() 
    24.    
  
  
  
  24. result = m.get_result_queue() 
    25.    
  
  
  
  25. # 從task隊(duì)列取任務(wù),并把結(jié)果寫入result隊(duì)列: 
    26.    
  
  
  
  26. for i in range(10): 
    27.    
  
  
  
  27.  try: 
    28.    
  
  
  
  28.  n = task.get(timeout=1) 
    29.    
  
  
  
  29.  print('run task %d * %d...' % (n, n)) 
    30.    
  
  
  
  30.  r = '%d * %d = %d' % (n, n, n*n) 
    31.    
  
  
  
  31.  time.sleep(1) 
    32.    
  
  
  
  32.  result.put(r) 
    33.    
  
  
  
  33.  except queue.Empty: 
    34.    
  
  
  
  34.  print('task queue is empty.') 
    35.    
  
  
  
  35. # 處理結(jié)束: 
    36.    
  
  
  
  36. print('worker exit.') 
    37.

先運(yùn)行task_master.py,然后再運(yùn)行task_worker.py

(1)task_master.py運(yùn)行結(jié)果如下

  
 
 
 
    
   
  
  
  
  1. start! 
    2.    
  
  
  
  2. Put task 7872... 
    3.    
  
  
  
  3. Put task 6931... 
    4.    
  
  
  
  4. Put task 1395... 
    5.    
  
  
  
  5. Put task 8477... 
    6.    
  
  
  
  6. Put task 8300... 
    7.    
  
  
  
  7. Put task 1597... 
    8.    
  
  
  
  8. Put task 8738... 
    9.    
  
  
  
  9. Put task 8627... 
    10.    
  
  
  
  10. Put task 1884... 
    11.    
  
  
  
  11. Put task 2561... 
    12.    
  
  
  
  12. Try get results... 
    13.    
  
  
  
  13. Result: 7872 * 7872 = 61968384 
    14.    
  
  
  
  14. Result: 6931 * 6931 = 48038761 
    15.    
  
  
  
  15. Result: 1395 * 1395 = 1946025 
    16.    
  
  
  
  16. Result: 8477 * 8477 = 71859529 
    17.    
  
  
  
  17. Result: 8300 * 8300 = 68890000 
    18.    
  
  
  
  18. Result: 1597 * 1597 = 2550409 
    19.    
  
  
  
  19. Result: 8738 * 8738 = 76352644 
    20.    
  
  
  
  20. Result: 8627 * 8627 = 74425129 
    21.    
  
  
  
  21. Result: 1884 * 1884 = 3549456 
    22.    
  
  
  
  22. Result: 2561 * 2561 = 6558721 
    23.    
  
  
  
  23. master exit. 
    24.

(2)task_worker.py運(yùn)行結(jié)果如下

  
 
 
 
    
   
  
  
  
  1. Connect to server 127.0.0.1... 
    2.    
  
  
  
  2. run task 8640 * 8640... 
    3.    
  
  
  
  3. run task 7418 * 7418... 
    4.    
  
  
  
  4. run task 9303 * 9303... 
    5.    
  
  
  
  5. run task 568 * 568... 
    6.    
  
  
  
  6. run task 1633 * 1633... 
    7.    
  
  
  
  7. run task 3583 * 3583... 
    8.    
  
  
  
  8. run task 3293 * 3293... 
    9.    
  
  
  
  9. run task 8975 * 8975... 
    10.    
  
  
  
  10. run task 8189 * 8189... 
    11.    
  
  
  
  11. run task 731 * 731... 
    12.    
  
  
  
  12. worker exit. 
    13.

知識(shí)補(bǔ)充

這個(gè)簡(jiǎn)單的Master/Worker模型有什么用?其實(shí)這就是一個(gè)簡(jiǎn)單但真正的分布式計(jì)算，把代碼稍加改造，啟動(dòng)多個(gè)worker，就可以把任務(wù)分布到幾臺(tái)甚至幾十臺(tái)機(jī)器上，比如把計(jì)算n*n的代碼換成發(fā)送郵件，就實(shí)現(xiàn)了郵件隊(duì)列的異步發(fā)送。

Queue對(duì)象存儲(chǔ)在哪?注意到task_worker.py中根本沒(méi)有創(chuàng)建Queue的代碼，所以，Queue對(duì)象存儲(chǔ)在task_master.py進(jìn)程中：

 

 

而Queue之所以能通過(guò)網(wǎng)絡(luò)訪問(wèn)，就是通過(guò)QueueManager實(shí)現(xiàn)的。由于QueueManager管理的不止一個(gè)Queue，所以，要給每個(gè)Queue的網(wǎng)絡(luò)調(diào)用接口起個(gè)名字，比如get_task_queue。task_worker這里的QueueManager注冊(cè)的名字必須和task_manager中的一樣。對(duì)比上面的例子，可以看出Queue對(duì)象從另一個(gè)進(jìn)程通過(guò)網(wǎng)絡(luò)傳遞了過(guò)來(lái)。只不過(guò)這里的傳遞和網(wǎng)絡(luò)通信由QueueManager完成。

authkey有什么用?這是為了保證兩臺(tái)機(jī)器正常通信，不被其他機(jī)器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。


網(wǎng)站欄目：Python分布式進(jìn)程中你會(huì)遇到的坑
本文來(lái)源：http://www.5511xx.com/article/dhgeioo.html