Python中的多线程技术

Question

在python中实现多线程时，我遇到了一些问题。这个问题对于我的用例来说非常具体。在同一个网站上浏览了许多帖子之后，我部署了最广泛建议/使用的方法。你知道吗

我从定义线程类开始，如下所示。你知道吗

class myThread(Thread):
    def __init__(self, graphobj, q):
        Thread.__init__(self)
        self.graphobj = graphobj
        self.q = q
    def run(self):
        improcess(self.graphobj, self.q)

我定义了我的函数来完成所有需要的处理。你知道吗

def improcess(graphobj, q):
    while not exitFlag:
        queueLock.acquire()
        if not q.empty():
            photo_id = q.get()
            queueLock.release()
            # Complete processing
        else:
            queueLock.release()

现在是我被困住的部分。我能够运行下面提到的代码完全是因为它没有任何问题。但是，如果我尝试将同一个函数包装成这样，它就会崩溃。你知道吗

def train_control(graphobj, photo_ids):
    workQueue = Queue(len(photo_ids))
    for i in range(1,5):
        thread = myThread(graphobj=graphobj, q=workQueue)
        thread.start()
        threads.append(thread)
    queueLock.acquire()
    for photo_id in photo_ids:
        workQueue.put(photo_id)
    queueLock.release()
    while not workQueue.empty():
        pass
    exitFlag = 1
    for t in threads:
        t.join()

我所说的中断是指线程完成了它们的工作，但它们并没有停止等待，即exitFlag从未设置为1。我不知道如何使这项工作。你知道吗

不幸的是，我们的系统的设计是这样的，这段代码需要被包装在一个可以被另一个模块调用的函数中，因此将其拉出并不是一个真正的选择。你知道吗

期待着专家们的意见。提前谢谢。你知道吗

编辑：初稿忘了提这个。我全局初始化exitFlag并将其值设置为0。你知道吗

下面是我创建的用于捕获此问题的最小可验证代码段：

import threading
import Queue

globvar01 = 5
globvar02 = 7
exitFlag = 0

globlist = []
threads = []

queueLock = threading.Lock()
workQueue = Queue.Queue(16)

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

def myfunc(threadID, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            thoughtnum = q.get()
            queueLock.release()
            print "Processing thread " + str(threadID)
            if (thoughtnum < globvar01):
                globlist.append([1,2,3])
            elif (thoughtnum < globvar02):
                globlist.append([2,3,4])
        else:
            queueLock.release()

def controlfunc():
    for i in range(1,5):
        thread = myThread(i, workQueue)
        thread.start()
        threads.append(thread)
    queueLock.acquire()
    for i in range(1,11):
        workQueue.put(i)
    queueLock.release()
    # Wait for queue to empty
    while not workQueue.empty():
        pass
    exitFlag = 1
    # Wait for all threads to complete
    for t in threads:
        t.join()

print "Starting main thread"

controlfunc()

print "Exiting Main Thread"

Answer 1

在生成任何线程之前，您需要确保将exitFlag设置为0（False），否则在impprocess()中它们将不会执行任何操作，并且队列将保持为非空。你知道吗

如果将exitFlag作为全局变量，并且没有从上一次运行中清除，则可能会发生此问题。你知道吗

Answer 2

从你的MCVE，唯一缺少的是：

while not workQueue.empty():
    pass
global exitFlag  # Need this or `exitFlag` is a local variable only.
exitFlag = 1

但是，您可以通过使用队列中的sentinel值来关闭工作线程，从而消除queueLock和exitFlag，并消除自旋等待。工作线程将在q.get()上休眠，并且主线程不必旋转等待空队列：

#!python2
from __future__ import print_function
import threading
import Queue

debug = 1
console = threading.Lock()

def tprint(*args,**kwargs):
    if debug:
        name = threading.current_thread().getName()
        with console:
            print('{}: '.format(name),end='')
            print(*args,**kwargs)

globvar01 = 5
globvar02 = 7

globlist = []
threads = []

workQueue = Queue.Queue(16)

class myThread(threading.Thread):
    def __init__(self, threadID, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.q = q
    def run(self):
        tprint("Starting thread " + str(self.threadID))
        myfunc(self.threadID, self.q)
        tprint("Exiting thread " + str(self.threadID))

def myfunc(threadID, q):
    while True:
        thoughtnum = q.get()
        tprint("Processing thread " + str(threadID))
        if thoughtnum is None:
            break
        elif thoughtnum < globvar01:
            globlist.append([1,2,3])
        elif thoughtnum < globvar02:
            globlist.append([2,3,4])

def controlfunc():
    for i in range(1,5):
        thread = myThread(i, workQueue)
        thread.start()
        threads.append(thread)
    for i in range(1,11):
        workQueue.put(i)
    # Wait for all threads to complete
    for t in threads:
        workQueue.put(None)
    for t in threads:
        t.join()

tprint("Starting main thread")

controlfunc()

tprint("Exiting Main Thread")

输出：

MainThread: Starting main thread
Thread-1: Starting thread 1
Thread-2: Starting thread 2
Thread-3: Starting thread 3
Thread-4: Starting thread 4
Thread-1: Processing thread 1
Thread-2: Processing thread 2
Thread-3: Processing thread 3
Thread-4: Processing thread 4
Thread-1: Processing thread 1
Thread-2: Processing thread 2
Thread-3: Processing thread 3
Thread-4: Processing thread 4
Thread-1: Processing thread 1
Thread-2: Processing thread 2
Thread-3: Processing thread 3
Thread-4: Processing thread 4
Thread-1: Processing thread 1
Thread-2: Processing thread 2
Thread-3: Exiting thread 3
Thread-4: Exiting thread 4
Thread-1: Exiting thread 1
Thread-2: Exiting thread 2
MainThread: Exiting Main Thread