欢迎各位兄弟 发布技术文章

这里的技术是共享的

You are here

首页

python初学分布式疑问，请pythoner帮忙看看，谢谢！

shiping1 的头像
星期五, 2014-09-26 22:23 — shiping1
最近通过廖雪峰的官网开始学习python  

http://www.liaoxuefeng.com/wiki/001374738125095c955c1e6d8bb493182103fac9270762a000/001386832973658c7...

python环境为2.7.8
学到分布式进程这一节运行服务端代码出现错误
PicklingError: Can't pickle <function <lambda> at 0x02747DB0>: it's not found as __main__.<lambda>

具体代码如下，大家也可以点击那个链接去查看
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# -*- coding: utf-8 -*-
# taskmanager.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', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey='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()
 
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# -*- coding: utf-8 -*-
# taskmanager.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', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey='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()
 
 
 
全部错误信息如下
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
Python 2.7.8 (default, Jun 30 2014, 16:03:49) [MSC v.1500 32 bit (Intel)] on win32
Type "copyright", "credits" or "license()" for more information.
>>> ================================ RESTART ================================
>>> 
 
Traceback (most recent call last):
  File "C:/Users/wangwei/Desktop/test", line 22, in <module>
    manager.start()
  File "C:\Python27\lib\multiprocessing\managers.py", line 524, in start
    self._process.start()
  File "C:\Python27\lib\multiprocessing\process.py", line 130, in start
    self._popen = Popen(self)
  File "C:\Python27\lib\multiprocessing\forking.py", line 277, in __init__
    dump(process_obj, to_child, HIGHEST_PROTOCOL)
  File "C:\Python27\lib\multiprocessing\forking.py", line 199, in dump
    ForkingPickler(file, protocol).dump(obj)
  File "C:\Python27\lib\pickle.py", line 224, in dump
    self.save(obj)
  File "C:\Python27\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Python27\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Python27\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 562, in save_tuple
    save(element)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Python27\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 562, in save_tuple
    save(element)
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 748, in save_global
    (obj, module, name))
PicklingError: Can't pickle <function <lambda> at 0x02747DB0>: it's not found as __main__.<lambda>
>>> 

 该代码在ubuntu python2.6.x上运行正常

请各位老手帮忙看下，谢谢！
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
回复次数：19
#1 得分：0 回复于： 2014-08-25 09:15:50
自己先顶顶。。。 
如果您对CSDN论坛有意见和建议 请直接在本帖指教
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
3
2
3
#2 得分：0 回复于： 2014-08-25 09:19:38
Python code?
1
2
3
# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())

# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())
这几行代码和教程中的代码不一样.

1. 试过教程中的代码吗?
2. 为什么做这个改动? 改动时参考register函数的文档了吗?
社区问答第11期：关东升 Swift开发指南
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#3 得分：0 回复于： 2014-08-25 09:24:03
引用 2 楼 panghuhu250 的回复:
Python code?
1
2
3
# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())

# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', Queue.Queue())
QueueManager.register('get_result_queue', lambda: Queue.Queue())
这几行代码和教程中的代码不一样.

1. 试过教程中的代码吗?
2. 为什么做这个改动? 改动时参考register函数的文档了吗?

非常感谢，这里我之前是按照教程上面代码写的，但发生错误后，我就试着改动了一下
贴代码时，忘了改回来，实在不好意思 
但出错信息是一样的
这里原始代码应该是：
Python code?
1
2
QueueManager.register('get_task_queue', lambda: task_queue)
QueueManager.register('get_result_queue', lambda: result_queue)

1
2
QueueManager.register('get_task_queue', lambda: task_queue)
QueueManager.register('get_result_queue', lambda: result_queue)

 
2015年1月微软MVP申请开始啦
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#4 得分：0 回复于： 2014-08-25 12:15:28
又没人了，，，， 自己再顶下。。。。 
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#5 得分：10 回复于： 2014-08-25 13:48:51
没有windows环境linux下直接运行脚本正常，那个错误看描述应该是在你的主脚本里没找到lambda表达式，看了下源码报这个错的代码是
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
try:
            __import__(module)
            mod = sys.modules[module]
            klass = getattr(mod, name)
        except (ImportError, KeyError, AttributeError):
            #这里抛出这个错
            raise PicklingError(
                "Can't pickle %r: it's not found as %s.%s" %
                (obj, module, name))
        else:
            if klass is not obj:
                raise PicklingError(
                    "Can't pickle %r: it's not the same object as %s.%s" %
                    (obj, module, name))

你找到你的pickle.py文件修改下，把
__import__(module)
mod = sys.modules[module]
klass = getattr(mod, name)
这3行代码复制到try上面一行，保存然后运行你自己的脚本看看到底是哪一行报错，修改之前记得先备份哦
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#6 得分：0 回复于： 2014-08-25 17:20:49
引用 5 楼 xydtf 的回复:
没有windows环境linux下直接运行脚本正常，那个错误看描述应该是在你的主脚本里没找到lambda表达式，看了下源码报这个错的代码是
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
try:
            __import__(module)
            mod = sys.modules[module]
            klass = getattr(mod, name)
        except (ImportError, KeyError, AttributeError):
            #这里抛出这个错
            raise PicklingError(
                "Can't pickle %r: it's not found as %s.%s" %
                (obj, module, name))
        else:
            if klass is not obj:
                raise PicklingError(
                    "Can't pickle %r: it's not the same object as %s.%s" %
                    (obj, module, name))

你找到你的pickle.py文件修改下，把
__import__(module)
mod = sys.modules[module]
klass = getattr(mod, name)
这3行代码复制到try上面一行，保存然后运行你自己的脚本看看到底是哪一行报错，修改之前记得先备份哦

非常感谢
在你的指点下，我修改pickle.py代码如下
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
if name is None:
            name = obj.__name__
 
        module = getattr(obj, "__module__", None)
        if module is None:
            module = whichmodule(obj, name)
        __import__(module)
        mod = sys.modules[module]
        klass = getattr(mod, name)
        try:
            pass
        except (ImportError, KeyError, AttributeError):
            raise PicklingError(
                "Can't pickle %r: it's not found as %s.%s" %
                (obj, module, name))


再次运行 出现错误 
Python code?
1
2
3
4
5
  File "C:\Python27\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Python27\lib\pickle.py", line 742, in save_global
    klass = getattr(mod, name)
AttributeError: 'module' object has no attribute '<lambda>'

 
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#7 得分：0 回复于： 2014-08-25 17:53:09
错误很明显，在__import__(module) 这行之上加一句
Python code?
1
print dir(module)


在你自己的脚本里加如下两个函数
Python code?
1
2
3
4
5
6
7
def get_task_queue():
    global task_queue
    return task_queue
 
def get_result_queue():
    global result_queue
    return result_queue


再把
Python code?
1
2
QueueManager.register('get_task_queue', lambda: task_queue)
QueueManager.register('get_result_queue', lambda: result_queue)

替换为
Python code?
1
2
QueueManager.register('get_task_queue', get_task_queue)
QueueManager.register('get_result_queue', get_result_queue)

如果还报找不到属性的错误，就看下开始加的print dir(module) 这句打印出来的信息判断下moudule到底是哪个模块
 
对我有用[1] 丢个板砖[0] 引用 | 举报 | 管理
#8 得分：0 回复于： 2014-08-25 17:57:09
打错了
print dir(module)
改为
print dir(mod),mod.__file__
 
对我有用[1] 丢个板砖[0] 引用 | 举报 | 管理
#9 得分：0 回复于： 2014-08-25 17:58:11
引用 8 楼 xydtf 的回复:
打错了
print dir(module)
改为
print dir(mod),mod.__file__

加在mod = sys.modules[module]
这行之后
 
对我有用[1] 丢个板砖[0] 引用 | 举报 | 管理
#10 得分：0 回复于： 2014-08-26 09:08:54
引用 9 楼 xydtf 的回复:
Quote: 引用 8 楼 xydtf 的回复:
打错了
print dir(module)
改为
print dir(mod),mod.__file__

加在mod = sys.modules[module]
这行之后
非常感谢你的耐心解答，根据你的方法修改后，现在没有出现找不到属性错误
但还是在manager.start()时出现了另外一个错误，如下：
Python code?
1
2
3
4
5
6
Traceback (most recent call last):
  File "D:\Python\taskmanager.py", line 37, in <module>
    manager.start()
  File "C:\Python27\lib\multiprocessing\managers.py", line 528, in start
    self._address = reader.recv()
EOFError

这个是不是还是在做pickle时发生的问题呢？ 
 
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#11 得分：0 回复于： 2014-08-26 13:48:03
像是pickle dump时不完整造成的，可能是多线程问题，put任务时加个锁试试
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#12 得分：0 回复于： 2014-08-27 09:18:53
引用 11 楼 xydtf 的回复:
像是pickle dump时不完整造成的，可能是多线程问题，put任务时加个锁试试

我把下面那一段全去了 只保留到
[code=python
]# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey='abc')
# 启动Queue:
manager.start()
[/code]
还是出现同样的错误  真郁闷    
Python code?
1
2
3
4
5
6
Traceback (most recent call last):
  File "D:\Python\taskmanager.py", line 37, in <module>
    manager.start()
  File "C:\Python27\lib\multiprocessing\managers.py", line 528, in start
    self._address = reader.recv()
EOFError


今天学习解析XML  
导包的时候竟然都出发问题了  
Python code?
1
2
3
4
5
6
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
sys.path.append('C:\\Python27\\Lib\\xml\\parsers\\expat.py')
print sys.path
from xml.parsers.expat import ParserCreate


错误如下：
Python code?
1
2
3
4
5
6
Traceback (most recent call last):
  File "D:\Python\xml.py", line 6, in <module>
    from xml.parsers.expat import ParserCreate
  File "D:\Python\xml.py", line 6, in <module>
    from xml.parsers.expat import ParserCreate
ImportError: No module named parsers.expat


不知道为什么。。。。。  太郁闷了 
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
2
#13 得分：10 回复于： 2014-08-27 09:28:44
这个是python内置的library 为什么不看官网的教程呢
https://docs.python.org/2/library/multiprocessing.html?highlight=basemanager#multiprocessing.manager...
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#14 得分：0 回复于： 2014-08-27 12:28:29
引用 13 楼 openXMPP 的回复:
这个是python内置的library 为什么不看官网的教程呢
https://docs.python.org/2/library/multiprocessing.html?highlight=basemanager#multiprocessing.manager...
非常感谢，对着官网上的教程实验可以成功，但原方法在linux上也是可以行得通的

官网教程在实验过程中发现一个问题
Python code?
1
2
3
4
5
6
7
8
from multiprocessing.managers import BaseManager
import Queue
queue = Queue.Queue()
lass QueueManager(BaseManager): pass
QueueManager.register('get_queue', callable=lambda:queue)
m = QueueManager(address=('', 50000), authkey='abracadabra')
s = m.get_server()
s.serve_forever()


Python code?
1
2
3
4
5
6
7
from multiprocessing.managers import BaseManager
class QueueManager(BaseManager): pass
QueueManager.register('get_queue')
m = QueueManager(address=('foo.bar.org', 50000), authkey='abracadabra')
m.connect()
queue = m.get_queue()
queue.put('hello')


这两段代码如果在同一台机器上运行会发生  下面这个错误 
Python code?
1
2
3
4
5
6
7
8
9
10
Traceback (most recent call last):
  File "D:/Python/task.py", line 20, in <module>
    m.connect()
  File "C:\Python27\lib\multiprocessing\managers.py", line 500, in connect
    conn = Client(self._address, authkey=self._authkey)
  File "C:\Python27\lib\multiprocessing\connection.py", line 175, in Client
    answer_challenge(c, authkey)
  File "C:\Python27\lib\multiprocessing\connection.py", line 428, in answer_challenge
    message = connection.recv_bytes(256)         # reject large message
EOFError


初学这个，请见谅，再次感谢！
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
2
#15 得分：0 回复于： 2014-08-27 12:34:11
'foo.bar.org', 50000
你本机有这个Domain吗 
换句话说 你ping这个'foo.bar.org' 地址 能到本机？
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#16 得分：0 回复于： 2014-08-27 12:37:20
引用 15 楼 openXMPP 的回复:
'foo.bar.org', 50000
你本机有这个Domain吗 
换句话说 你ping这个'foo.bar.org' 地址 能到本机？

这个Domain我已改成本机IP地址了  上面的代码是直接复制官网教程的，没有改过来，不好意思  
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#17 得分：0 回复于： 2014-08-27 12:41:17
引用 15 楼 openXMPP 的回复:
'foo.bar.org', 50000
你本机有这个Domain吗 
换句话说 你ping这个'foo.bar.org' 地址 能到本机？

代码如下
Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#!/usr/bin/env python
# -*- coding: utf-8 -*-
 
# task.py
 
import random, time, Queue
from multiprocessing.managers import BaseManager
 
class QueueManager(BaseManager):
    pass
    
QueueManager.register('get_task_queue')
 
server_addr = '10.135.90.95'
print('Connect to server %s ...' % server_addr)
 
m = QueueManager(address=(server_addr, 50000), authkey='abc')
m.connect()
 
# 获得通过网络访问的Queue对象
task = m.get_task_queue()
# 加入任务
task.put('hello')



Python code?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# taskmanager.py
 
import random, time, Queue, pickle
from multiprocessing.managers import BaseManager
 
# 发送任务队列
task_queue = Queue.Queue();
 
# 从BaseManager继承QueueManager
class QueueManager(BaseManager):
    pass
    
# 将两个队列注册到网络上,callable参数关联Queue对象
QueueManager.register('get_task_queue',  callable=lambda: task_queue)
# 绑定端口5000， 设置验证码'abc'
manager = QueueManager(address=('', 50000), authkey='abc')
# 启动Queue
s = manager.get_server()
s.serve_forever()
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
2
#18 得分：0 回复于： 2014-08-27 14:05:33
把你的代码粘贴复制 又运行 没出现你说的问题 
 
对我有用[0] 丢个板砖[0] 引用 | 举报 | 管理
#19 得分：0 回复于： 2014-08-27 14:57:38
引用 18 楼 openXMPP 的回复:
把你的代码粘贴复制 又运行 没出现你说的问题 
找到原因了
我用Python IDLE运行的，两个代码都在一个shell里运行时发生了这个错误 
开多个shell 时 运行正常，再次感谢！

普通分类: 
python
Powered by Drupal

友情链接

校园好文网

 www.shipingzhong.cn  个人技术网_前端_后台    备案号:苏ICP备18010659号-2