[python](windows)分布式進程問題：pickle模塊不能序列化lambda函數

運行錯誤：_pickle.PicklingError: Can't pickle <function <lambda> at 0x000002BAAEF12F28>: attribute lookup <lambda> on __main__ failed

代碼如下：

#!/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

# 把兩個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.')

報錯信息：

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

錯誤原因：pickle模塊不能序列化lambda，需要自定義函數

修改代碼如下：

 

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

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

# 發送任務的隊列:
task_queue = queue.Queue()
# 接收結果的隊列:
result_queue = queue.Queue()

# 自定義函數re_task_queue
def re_task_queue():  
    global task_queue  
    return task_queue  

# 自定義函數re_result_queue
def re_result_queue():  
    global result_queue  
    return result_queue

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

if __name__ == '__main__':

    # 把兩個Queue都注冊到網絡上, callable參數關聯了Queue對象:
    QueueManager.register('get_task_queue', callable=re_task_queue)
    QueueManager.register('get_result_queue', callable=re_result_queue)

    # 綁定端口5000, 設置驗證碼'abc':
    manager = QueueManager(address=('127.0.0.1', 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.')

 

運行結果：

C:\Users\Lucky丶M\python>python task_master.py
Put task 4962...
Put task 3460...
Put task 4774...
Put task 4301...
Put task 9120...
Put task 7183...
Put task 4915...
Put task 3173...
Put task 9138...
Put task 5798...
Try get results...