如何让Python线程优雅地完成

Question

我正在做一个涉及数据收集和日志记录的项目。我有两个线程在运行，一个收集线程和一个日志线程，都是在main中启动的。我正试图允许程序在使用Ctrl-C时正常终止

我正在使用threading.Event向线程发送信号，以结束它们各自的循环。停止sim_collectData方法可以很好地工作，但是它似乎没有正确地停止logData线程。print语句从不执行，程序只是暂停。（它没有结束，只是坐在那里）。

logData中的第二个while循环是确保队列中的所有内容都被记录下来。目标是让Ctrl-C立即停止收集线程，然后让日志线程完成清空队列，然后才完全终止程序。（现在，数据只是被打印出来——最终会被记录到数据库中）。

我不明白为什么第二个线程永远不会终止。我所做的一切都是基于这个答案：Stopping a thread after a certain amount of time。我错过了什么？

def sim_collectData(input_queue, stop_event):
    ''' this provides some output simulating the serial
    data from the data logging hardware. 
    '''
    n = 0
    while not stop_event.is_set():
        input_queue.put("DATA: <here are some random data> " + str(n))
        stop_event.wait(random.randint(0,5))
        n += 1
    print "Terminating data collection..."
    return

def logData(input_queue, stop_event):
    n = 0

    # we *don't* want to loop based on queue size because the queue could
    # theoretically be empty while waiting on some data.
    while not stop_event.is_set():
        d = input_queue.get()
        if d.startswith("DATA:"):
            print d
        input_queue.task_done()
        n += 1

    # if the stop event is recieved and the previous loop terminates, 
    # finish logging the rest of the items in the queue.
    print "Collection terminated. Logging remaining data to database..."
    while not input_queue.empty():
        d = input_queue.get()
        if d.startswith("DATA:"):
            print d
        input_queue.task_done()
        n += 1
    return


def main():
    input_queue = Queue.Queue()

    stop_event = threading.Event() # used to signal termination to the threads

    print "Starting data collection thread...",
    collection_thread = threading.Thread(target=sim_collectData, args=(input_queue,     stop_event))
    collection_thread.start()
    print "Done."

    print "Starting logging thread...",
    logging_thread = threading.Thread(target=logData, args=(input_queue, stop_event))
    logging_thread.start()
    print "Done."

    try:
        while True:
        time.sleep(10)
    except (keyboardInterrupt, SystemExit):
        # stop data collection. Let the logging thread finish logging everything in the queue
        stop_event.set()

 main()

Answer 1

问题是日志记录器正在等待d = input_queue.get()，并且不会检查事件。一种解决方案是完全跳过事件，并创建一个唯一的消息，告诉记录器停止。收到信号后，将该消息发送到队列。

import threading
import Queue
import random
import time

def sim_collectData(input_queue, stop_event):
    ''' this provides some output simulating the serial
    data from the data logging hardware. 
    '''
    n = 0
    while not stop_event.is_set():
        input_queue.put("DATA: <here are some random data> " + str(n))
        stop_event.wait(random.randint(0,5))
        n += 1
    print "Terminating data collection..."
    input_queue.put(None)
    return

def logData(input_queue):
    n = 0

    # we *don't* want to loop based on queue size because the queue could
    # theoretically be empty while waiting on some data.
    while True:
        d = input_queue.get()
        if d is None:
            input_queue.task_done()
            return
        if d.startswith("DATA:"):
            print d
        input_queue.task_done()
        n += 1

def main():
    input_queue = Queue.Queue()

    stop_event = threading.Event() # used to signal termination to the threads

    print "Starting data collection thread...",
    collection_thread = threading.Thread(target=sim_collectData, args=(input_queue,     stop_event))
    collection_thread.start()
    print "Done."

    print "Starting logging thread...",
    logging_thread = threading.Thread(target=logData, args=(input_queue,))
    logging_thread.start()
    print "Done."

    try:
        while True:
            time.sleep(10)
    except (KeyboardInterrupt, SystemExit):
        # stop data collection. Let the logging thread finish logging everything in the queue
        stop_event.set()

main()

Answer 2

您正在对您的input_queue调用阻塞get，没有超时。在logData的任一部分中，如果调用input_queue.get()并且队列为空，则它将无限期阻塞，从而阻止logging_thread到达完成。

要修复此问题，您需要调用input_queue.get_nowait()或将超时传递给input_queue.get()。

我的建议是：

def logData(input_queue, stop_event):
    n = 0

    while not stop_event.is_set():
        try:
            d = input_queue.get_nowait()
            if d.startswith("DATA:"):
                print "LOG: " + d
                n += 1
        except Queue.Empty:
            time.sleep(1)
    return

您还向线程发出终止的信号，但不是等待它们这样做。考虑在main函数中执行此操作。

try:
    while True:
        time.sleep(10)
except (KeyboardInterrupt, SystemExit):
    stop_event.set()
    collection_thread.join()
    logging_thread.join()

Answer 3

我不是线程专家，但在您的logData函数中，第一个d=input_queue.get()是阻塞的，即，如果队列是空的，它将永远等待，直到收到队列消息。这可能就是为什么logData线程从未终止，它一直在等待队列消息。

请参阅[Python docs]将其更改为非阻塞队列读取：使用.get(False)或.get_nowait()-但在队列为空的情况下，两者都需要一些异常处理。