python - 使用 `thread.join()` 时多线程冻结
问题描述
我正在尝试设置 3 个线程并在队列中执行 5 个任务。这个想法是线程将首先同时运行前 3 个任务,然后 2 个线程完成剩下的 2 个。但是程序似乎冻结了。我没有发现它有什么问题。
from multiprocessing import Manager
import threading
import time
global exitFlag
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):
global exitFlag
while not exitFlag:
if not workQueue.empty():
data = q.get()
print("%s processing %s" % (threadName, data))
else:
pass
time.sleep(1)
print('Nothing to Process')
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Manager().Queue(10)
threads = []
threadID = 1
# create thread
for tName in threadList:
thread = myThread(threadID, tName, workQueue)
thread.start()
threads.append(thread)
threadID += 1
# fill up queue
queueLock.acquire()
for word in nameList:
workQueue.put(word)
queueLock.release()
# wait queue clear
while not workQueue.empty():
pass
# notify thread exit
exitFlag = 1
# wait for all threads to finish
for t in threads:
t.join()
print("Exiting Main Thread")
我不知道究竟发生了什么,但是在我删除该join()
部分后,程序能够运行得很好玩。我不明白的是,exitFlag 应该在队列清空时发出信号。因此,process_data() 似乎以某种方式未检测到信号
解决方案
您的代码存在多个问题。首先,由于全局解释器锁 ( GIL ) ,CPython 中的线程不会“同时”运行 Python 代码。线程必须持有 GIL 才能执行 Python 字节码。默认情况下,一个线程将 GIL 保存最多 5 毫秒(Python 3.2+),如果它没有提前删除它,因为它确实阻塞了 I/O。对于 Python 代码的并行执行,您必须使用multiprocessing
.
您还不必要地使用 aManager.Queue
而不是 a queue.Queue
。AManager.Queue
是queue.Queue
一个单独的管理器进程。您在这里引入了 IPC 和内存复制的绕道,但没有任何好处。
死锁的原因是你在这里有一个竞争条件:
if not workQueue.empty():
data = q.get()
这不是原子操作。一个线程可以检查workQueue.empty()
,然后删除 GIL,让另一个线程排空队列,然后继续执行data = q.get()
,如果您不再将某些内容放入队列,这将永远阻塞。 Queue.empty()
检查是一般的反模式,没有必要使用它。使用毒丸(哨兵值)来打破 get-loop 并让工人知道他们应该退出。您需要与工人一样多的哨兵值。iter(callabel, sentinel)
在此处查找更多信息。
import time
from queue import Queue
from datetime import datetime
from threading import Thread, current_thread
SENTINEL = 'SENTINEL'
class myThread(Thread):
def __init__(self, func, inqueue):
super().__init__()
self.func = func
self._inqueue = inqueue
def run(self):
print(f"{datetime.now()} {current_thread().name} starting")
self.func(self._inqueue)
print(f"{datetime.now()} {current_thread().name} exiting")
def process_data(_inqueue):
for data in iter(_inqueue.get, SENTINEL):
print(f"{datetime.now()} {current_thread().name} "
f"processing {data}")
time.sleep(1)
if __name__ == '__main__':
N_WORKERS = 3
inqueue = Queue()
input_data = ["One", "Two", "Three", "Four", "Five"]
sentinels = [SENTINEL] * N_WORKERS # one sentinel value per worker
# enqueue input and sentinels
for word in input_data + sentinels:
inqueue.put(word)
threads = [myThread(process_data, inqueue) for _ in range(N_WORKERS)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"{datetime.now()} {current_thread().name} exiting")
示例输出:
2019-02-14 17:58:18.265208 Thread-1 starting
2019-02-14 17:58:18.265277 Thread-1 processing One
2019-02-14 17:58:18.265472 Thread-2 starting
2019-02-14 17:58:18.265542 Thread-2 processing Two
2019-02-14 17:58:18.265691 Thread-3 starting
2019-02-14 17:58:18.265793 Thread-3 processing Three
2019-02-14 17:58:19.266417 Thread-1 processing Four
2019-02-14 17:58:19.266632 Thread-2 processing Five
2019-02-14 17:58:19.266767 Thread-3 exiting
2019-02-14 17:58:20.267588 Thread-1 exiting
2019-02-14 17:58:20.267861 Thread-2 exiting
2019-02-14 17:58:20.267994 MainThread exiting
Process finished with exit code 0
如果你不坚持子类化Thread
,你也可以只使用multiprocessing.pool.ThreadPool
akamultiprocessing.dummy.Pool
在后台为你做管道。
推荐阅读
- python-3.x - Celery 真的是异步的吗?
- python - 如何将字符串格式应用于 bash 命令(通过子进程合并到 Python 脚本中)?
- dataframe - 如何根据每月的一周将日期截断为星期五?
- php - 输出与 sql 查询匹配的所有行
- r - 使用列表作为 dplyr::case_when 的结果时丢失名称
- rabbitmq - 暂停 MessageListenerContainer 以在指定的时间间隔内使用新消息
- app-store - 我可以将 App Store 的年龄更改为 18 岁以上吗?
- ruby - 如何在黄瓜红宝石中找到方法的源位置?
- css - 为什么 window.print 给出重复页面?
- c++ - 如何终止通过 CreateDesktop() 创建的桌面和其中的所有程序?