Python线程:我缺少什么？(task_done()调用次数太多)

Question

我为前面的长篇大论道歉.希望它能为解决方案提供足够的背景信息.我试图创建一个实用程序函数,它将使用任意数量的旧函数并将classmethod它们粘贴到多线程队列中:

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

这个错误在每个第二或第三个请求的随机线程中出现,因此很难准确确定问题所在.

任何人有任何想法和/或建议？

谢谢.

---

编辑:

我添加了prints以试图调试它(快速而肮脏而不是记录).在调用()之前print 'running thread: %s' % self.name的第一行run和另一行的一个print语句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

Answer 1

你解决这个问题的方法是"非常规的".但是现在忽略了这个问题......问题只是在你给出的代码中

q.put(QueuedCall('twitter', q, Twitter.get_status, [5,], __op_complete))

显然可以进行以下工作流程

1. 线程由QueuedCall构建并启动.__ init__
2. 然后将其放入队列q中.但是......在Queue完成其插入项的逻辑之前,独立线程已经完成其工作并尝试调用q.task_done().这会导致您的错误(在对象安全地放入队列之前调用了task_done())

怎么做？您不会将线程插入队列.队列保存线程处理的数据.相反,你

• 创建一个队列.插入你想要完成的工作(例如函数,他们想要的args和回调)
• 您创建并启动工作线程
• 工作线程调用
  • q.get()获取要调用的函数
  • 调用它
  • 调用q.task_done()让队列知道项目已被处理.