python - 当我放入队列或从队列中取出队列时，我是否应该麻烦锁定队列？

Question

我一直在学习有关 python3 中的多线程和队列的教程。作为official tutorial继续说，“这个模块中的 Queue 类实现了所有必需的锁定语义”。但是在another tutorial ，我见过一个例子如下:

import queue
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
   def __init__(self, threadID, name, q):
      threading.Thread.__init__(self)
      self.threadID = threadID
      self.name = name
      self.q = q
   def run(self):
      print ("Starting " + self.name)
      process_data(self.name, self.q)
      print ("Exiting " + self.name)

def process_data(threadName, q):
   while not exitFlag:
      queueLock.acquire()
      if not workQueue.empty():
         data = q.get()
         queueLock.release()
         print ("%s processing %s" % (threadName, data))
      else:
         queueLock.release()
         time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1

# Create new threads
for tName in threadList:
   thread = myThread(threadID, tName, workQueue)
   thread.start()
   threads.append(thread)
   threadID += 1

# Fill the queue
queueLock.acquire()
for word in nameList:
   workQueue.put(word)
queueLock.release()

# Wait for queue to empty
while not workQueue.empty():
   pass

# Notify threads it's time to exit
exitFlag = 1

# Wait for all threads to complete
for t in threads:
   t.join()
print ("Exiting Main Thread")

Answer 1

我相信您正在学习的教程是如何使用 Python 的线程安全队列的一个糟糕示例。特别是，本教程以一种不幸需要额外锁的方式使用线程安全队列。实际上，这个额外的锁意味着教程中的线程安全队列可以替换为基于简单列表的老式非线程安全队列。

Queue.empty() 的文档暗示了需要锁定的原因。 :

If empty() returns False it doesn't guarantee that a subsequent call to get() will not block.

问题是另一个线程可能在调用 empty() 和调用 get() 之间运行，窃取 empty() 否则报告存在的项目。教程可能使用锁来确保线程从调用 empty() 到调用 get() 期间对队列具有独占访问权限。如果没有此锁，两个线程可能会进入 if 语句并同时发出对 get() 的调用，这意味着其中一个可能会阻塞，等待一个永远不会被推送的项目。

---

让我向您展示如何正确使用线程安全队列。与其先检查 empty()，不如直接依赖 get() 的阻塞行为:

def process_data(threadName, q):
    while True:
        data = q.get()
        if exitFlag:
            break
        print("%s processing %s" % (threadName, data))

队列的内部锁定将确保在调用 get() 期间两个线程不会干扰，并且不需要 queueLock。请注意，本教程的原始代码将每 1 秒定期检查 exitFlag，而此修改后的队列要求您在将 exitFlag 设置为 True 后将虚拟对象插入队列——否则，将永远不会检查该标志.

Controller 代码的最后一部分需要修改如下:

# Notify threads it's time to exit
exitFlag = 1
for _ in range(len(threadList)):
    # Push a dummy element causing a single thread to wake-up and stop.
    workQueue.put(None)
# Wait for all threads to exit
for t in threads:
    t.join()

---

本教程使用线程安全队列还有一个问题，即在等待队列清空时在主线程中使用了忙循环:

# Wait for queue to empty
while not workQueue.empty():
    pass

等待队列清空最好使用Queue.task_done()在线程中然后调用 Queue.join()在主线程中。在 process_data() 循环体的末尾，调用 q.task_done()。在主 Controller 代码中，调用 q.join() 而不是上面的 while 循环。

另见 example在队列模块的 Python 文档页面底部。

---

或者，您可以保留 queueLock 并将线程安全队列替换为普通的旧列表，如下所示:

• 将 workQueue = queue.Queue(10) 替换为 workQueue = []
• 将 if not workQueue.empty() 替换为 if len(workQueue) > 0
• 将 workQueue.get() 替换为 workQueue.pop(0)
• 将 workQueue.put(word) 替换为 workQueue.append(word)

请注意，这不会保留原始版本中 put() 的阻塞行为。