Python：如何同时运行进度条和程序？

Question

我想知道如何在Python(2.7.x)中同时运行进度条和其他任务，然后在任务完成时停止进度条。

import sys, time
def progress_bar():
 while True:
  for c in ['-','\\','|','/']:
   sys.stdout.write('\r' + "Working " + c)
   sys.stdout.flush()
   time.sleep(0.2)

def work():
 *doing hard work*

我该如何实现类似以下功能：

progress_bar() #run in background?
work()
*stop progress bar*
print "\nThe work is done!"

- DanielTA

---

实际上，我会把它简单地称为“旋转器”，而不是进度条。 - martineau

@martineau 现在我想起来了，我同意。 - DanielTA

Answer 1

您可以使用threading模块在后台运行一个线程。例如：

def run_progress_bar(finished_event):
    chars = itertools.cycle(r'-\|/')
    while not finished_event.is_set():
        sys.stdout.write('\rWorking ' + next(chars))
        sys.stdout.flush()
        finished_event.wait(0.2)


# somewhere else...
finished_event = threading.Event()
progress_bar_thread = threading.Thread(target=run_progress_bar, args=(finished_event,))
progress_bar_thread.start()
# do stuff
finished_event.set()
progress_bar_thread.join()

- icktoofay

---

谢谢！不过有一个离题的问题。使用itertools在性能上比使用列表（像我所做的）更快吗？ - DanielTA

“progress_bar_thread.join()”这行代码是做什么用的？它似乎在有或没有它的情况下都能正常工作。 - DanielTA

@DanielTA：我使用了itertools，因为我认为它更加优雅，并且很好地适应了while循环；我不知道它是否更有效率，但无论如何，这都不会是您程序的瓶颈（# do stuff部分才是！），所以您不必担心。progress_bar_thread.join()行确保该线程已退出，因此在您在主线程上打印其他信息时，它不会再尝试打印另一个“Working”行。 - icktoofay

线程包是Python3标准库的一部分。因此，只需使用import threading即可。 - Prometheus

Answer 2

您可以创建一个单独的线程来显示进度条。这可以像@icktoofay的答案所示那样完成，但我更喜欢以下实现方式，它派生了一个新的线程子类来完成任务。这种方法的一个优点是，在每个新类的实例中，所有内容都是自包含的，因此您不需要全局变量来进行它们和主线程之间的通信。

import sys
import threading
import time

class ProgressBarThread(threading.Thread):
    def __init__(self, label='Working', delay=0.2):
        super(ProgressBarThread, self).__init__()
        self.label = label
        self.delay = delay  # interval between updates
        self.running = False
    def start(self):
        self.running = True
        super(ProgressBarThread, self).start()
    def run(self):
        label = '\r' + self.label + ' '
        while self.running:
            for c in ('-', '\\', '|', '/'):
                sys.stdout.write(label + c)
                sys.stdout.flush()
                time.sleep(self.delay)
    def stop(self):
        self.running = False
        self.join()  # wait for run() method to terminate
        sys.stdout.write('\r' + len(self.label)*' ' + '\r')  # clean-up
        sys.stdout.flush()

def work():
    time.sleep(5)  # *doing hard work*

pb_thread = ProgressBarThread('Computing')
pb_thread.start()
work()
pb_thread.stop()
print("The work is done!")

- martineau