对于前面的长篇帖子,我深表歉意。希望它能为解决方案提供足够的上下文。我试图创建一个实用程序函数,它将采用任意数量的旧 classmethod
s 并将它们放入多线程队列中:
class QueuedCall(threading.Thread):
def __init__(self, name, queue, fn, args, cb):
threading.Thread.__init__(self)
self.name = name
self._cb = cb
self._fn = fn
self._queue = queue
self._args = args
self.daemon = True
self.start()
def run(self):
r = self._fn(*self._args) if self._args is not None \
else self._fn()
if self._cb is not None:
self._cb(self.name, r)
self._queue.task_done()
这是我的调用代码的样子(在一个类中)
data = {}
def __op_complete(name, r):
data[name] = r
q = Queue.Queue()
socket.setdefaulttimeout(5)
q.put(QueuedCall('twitter', q, Twitter.get_status, [5,], __op_complete))
q.put(QueuedCall('so_answers', q, StackExchange.get_answers,
['api.stackoverflow.com', 534476, 5], __op_complete))
q.put(QueuedCall('so_user', q, StackExchange.get_user_info,
['api.stackoverflow.com', 534476], __op_complete))
q.put(QueuedCall('p_answers', q, StackExchange.get_answers,
['api.programmers.stackexchange.com', 23901, 5], __op_complete))
q.put(QueuedCall('p_user', q, StackExchange.get_user_info,
['api.programmers.stackexchange.com', 23901], __op_complete))
q.put(QueuedCall('fb_image', q, Facebook.get_latest_picture, None, __op_complete))
q.join()
return data
我在这里遇到的问题是,它似乎每次 都在新服务器重新启动时工作,但每隔第二或第三次请求就会失败,并出现错误:
ValueError: task_done() called too many times
此错误每隔一秒或三次请求就会出现在一个随机线程中,因此很难确切地确定问题出在哪里。
有人有任何想法和/或建议吗?
谢谢。
编辑:
我添加了 print
这是为了调试它(快速而肮脏而不是日志记录)。在 print 'running thread: %s' % self.name
的第一行打印一条语句( run
)和另一个在打电话之前的权利task_done()
(print 'thread done: %s' % self.name
)。
成功请求的输出:
running thread: twitter
running thread: so_answers
running thread: so_user
running thread: p_answers
thread done: twitter
thread done: so_user
running thread: p_user
thread done: so_answers
running thread: fb_image
thread done: p_answers
thread done: p_user
thread done: fb_image
不成功请求的输出:
running thread: twitter
running thread: so_answers
thread done: twitter
thread done: so_answers
running thread: so_user
thread done: so_user
running thread: p_answers
thread done: p_answers
Exception in thread p_answers:
Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 552, in __bootstrap_inner
self.run()
File "/home/demian/src/www/projects/demianbrecht/demianbrecht/demianbrecht/helpers.py", line 37, in run
self._queue.task_done()
File "/usr/lib/python2.7/Queue.py", line 64, in task_done
raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times
running thread: p_user
thread done: p_user
running thread: fb_image
thread done: fb_image
最佳答案
您解决这个问题的方法是“非常规的”。但是现在忽略它......问题只是在你给出的代码中
q.put(QueuedCall('twitter', q, Twitter.get_status, [5,], __op_complete))
显然有可能发生以下工作流
- 线程由QueuedCall.__init__构造并启动
- 然后放入队列q。但是......在队列完成其插入项目的逻辑之前,独立线程已经完成其工作并试图调用 q.task_done()。这会导致您遇到错误(在对象安全放入队列之前已调用 task_done())
应该怎么做?您不会将线程插入队列。队列保存线程处理的数据。所以相反你
- 创建队列。将你想要完成的工作插入其中(例如函数、他们想要的参数和回调)
- 您创建并启动工作线程
- 工作线程调用
- q.get() 获取要调用的函数
- 调用它
- 调用 q.task_done() 让队列知道项目已处理。
关于Python 线程 : What am I missing?(task_done() 调用次数过多),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/7439401/