我正在尝试设置3个线程并在队列中执行5个任务.这个想法是线程将首先同时运行前三个任务,然后两个线程将完成其余两个任务.但是该程序似乎冻结了.我无法发现任何问题.
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字节码.默认情况下,如果线程没有阻止它,因为它确实阻塞了I / O,所以它在GIL中最多保留5毫秒(Python 3.2).为了并行执行Python代码,您将必须使用多重处理.
您也可以不必要地使用Manager.Queue而不是queue.Queue. Manager.Queue是在单独的管理器进程上的queue.Queue.您在此处引入了绕过IPC和内存复制的弯路,但没有任何好处.
造成僵局的原因是您在这里有竞争状况：
if not workQueue.empty():
data = q.get()
这不是原子操作.线程可以检查workQueue.empty(),然后删除GIL,让另一个线程耗尽队列,然后继续执行data = q.get(),如果您不将某些内容再次放入队列,它将永远阻塞. Queue.empty()检查是一种常规的反模式,不需要使用它.使用有毒药丸(前哨值)来代替获取循环并让工人知道他们应该退出.您需要与工作人员一样多的哨兵值.查找有关iter(callabel,sentinel)here的更多信息.
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","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也就是multiprocessing.dummy.Pool,它在后台为您做管道.