【问题标题】:Is my HelloWorld queue working?我的 HelloWorld 队列是否正常工作?
【发布时间】:2011-08-24 02:25:24
【问题描述】:

我即将在应用程序中使用此设计,但我对 python 中的线程和队列内容相当陌生。显然,实际的应用程序不是用来打招呼的,但设计是一样的——即有一个过程需要一些时间来设置和拆除,但我可以一次完成多项任务。任务将随机到达,而且通常是突发的。

这是一个明智且线程安全的设计吗?

class HelloThing(object):

  def __init__(self):
    self.queue = self._create_worker()

  def _create_worker(self):
    import threading, Queue

    def worker():
      while True:
        things = [q.get()]
        while True:
          try:
            things.append(q.get_nowait())
          except Queue.Empty:
            break
        self._say_hello(things)
        [q.task_done() for task in xrange(len(things))]

    q = Queue.Queue()
    n_worker_threads = 1
    for i in xrange(n_worker_threads):
      t = threading.Thread(target=worker)
      t.daemon = True
      t.start()

    return q

  def _say_hello(self, greeting_list):
    import time, sys
    # setup stuff
    time.sleep(1)
    # do some things
    sys.stdout.write('hello {0}!\n'.format(', '.join(greeting_list)))
    # tear down stuff
    time.sleep(1)


if __name__ == '__main__':
  print 'enter __main__'

  import time
  hello = HelloThing()

  hello.queue.put('world')
  hello.queue.put('cruel world')
  hello.queue.put('stack overflow')

  time.sleep(2)

  hello.queue.put('a')
  hello.queue.put('b')

  time.sleep(2)

  for i in xrange(20):
    hello.queue.put(str(i))

  #hello.queue.join()

  print 'finish __main__'

【问题讨论】:

    标签: python multithreading queue


    【解决方案1】:
    1. 线程安全由 Queue 实现处理(如果需要,您也必须在 _say_hello 实现中处理)。

    2. 突发处理程序问题:突发应仅由单个线程处理。(例如:假设您的进程设置/拆卸需要 10 秒;在第 1 秒,所有线程都将忙于从第 0 秒开始的突发)第二个 5 一个新任务(或突发),但没有可用的线程来处理它们/它)。因此,突发应该由特定时间窗口的最大任务数(或者可能是“无限”)来定义。队列中的条目应该是一个任务列表。

    如何对突发任务列表进行分组? 我以代码的形式提供解决方案,更容易解释...

    producer_q = Queue()
    def _burst_thread():
       while True:
          available_tasks = [producer_q.get()]
          time.sleep(BURST_TIME_WINDOW)
          available_tasks.extend(producer_q.get() # I'm the single consumer, so will be at least qsize elements  
                                 for i in range(producer_q.qsize()))
          consumer_q.push(available_tasks)
    

    如果你想在一个突发中拥有最多的消息,你只需要将 available_tasks 分割到多个列表中。

    【讨论】:

    • 您的意思是用extend 代替expand 吗? qsize 的文档字符串也表示它是近似的,不可靠,所以在 q.get_nowait() 上使用 try:except: 会不会更好,就像我原来的例子一样?
    • @wim 是的,扩展是正确的方法。 qsize() 文档字符串中的“近似”表示不可靠(如果另一个并发线程获取或放置元素,后续调用可能会返回不同的内容);在我的示例中,并发线程只能放入新元素,因此 get 不会被阻塞。你的建议也可以,但有一个细微的差别(我的 - 在 BURST_TIME_WINDOW 过期后接受所有可用的任务,你的 - 接受所有 + 所有即将到来的任务,直到 producer_q 变空)
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2020-10-28
    • 2021-08-30
    • 2016-02-25
    • 2018-06-03
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多