一篇文章读懂 Python 多线程

Threading模块从 Python 1.5.2 版开始出现，用于增强底层的多线程模块thread。Threading 模块让操作多线程变得更简单，并且支持程序同时运行多个操作。

注意，Python 中的多线程最好用于处理有关 I/O 的操作，如从网上下载资源或者从本地读取文件或者目录。如果你要做的是 CPU 密集型操作，那么你需要使用 Python 的multiprocessing模块。这样做的原因是，Python 有一个全局解释器锁 (GIL)，使得所有子线程都必须运行在同一个主线程中。正因为如此，当你通过多线程来处理多个 CPU 密集型任务时，你会发现它实际上运行的更慢。因此，我们将重点放在那些多线程最擅长的领域：I/O 操作!

线程简介

多线程能让你像运行一个独立的程序一样运行一段长代码。这有点像调用子进程(subprocess)，不过区别是你调用的是一个函数或者一个类，而不是独立的程序。在我看来，举例说明更有助于解释。下面来看一个简单的例子：

1. import threading 
2.  
3. def doubler(number): 
4. """ 
5. 可以被线程使用的一个函数 
6. """ 
7. print(threading.currentThread.getName + '\n') 
8. print(number * 2) 
9. print 
10.  
11. if __name__ == '__main__': 
12. for i in range(5): 
13. my_thread = threading.Thread(target=doubler, args=(i,)) 
14. my_thread.start 

这里，我们导入 threading 模块并且创建一个叫 doubler的常规函数。这个函数接受一个值，然后把这个值翻一番。它还会打印出调用这个函数的线程的名称，并在最后打印一行空行。然后在代码的最后一块，我们创建五个线程并且依次启动它们。在我们实例化一个线程时，你会注意到，我们把 doubler 函数传给target参数，同时也给 doubler 函数传递了参数。Args参数看起来有些奇怪，那是因为我们需要传递一个序列给 doubler 函数，但它只接受一个变量，所以我们把逗号放在尾部来创建只有一个参数的序列。

需要注意的是，如果你想等待一个线程结束，那么需要调用 join方法。

当你运行以上这段代码，会得到以下输出内容：

1. Thread-1 
2.  
3. 0 
4.  
5. Thread-2 
6.  
7. 2 
8.  
9. Thread-3 
10.  
11. 4 
12.  
13. Thread-4 
14.  
15. 6 
16.  
17. Thread-5 
18.  
19. 8 

当然，通常情况下你不会希望输出打印到标准输出。如果不幸真的这么做了，那么最终的显示效果将会非常混乱。你应该使用 Python 的 logging 模块。它是线程安全的，并且表现出色。让我们用 logging模块修改上面的例子并且给我们的线程命名。代码如下：

1. import logging 
2. import threading 
3.  
4. def get_logger: 
5. logger = logging.getLogger("threading_example") 
6. logger.setLevel(logging.DEBUG) 
7.  
8. fh = logging.FileHandler("threading.log") 
9. fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s' 
10. formatter = logging.Formatter(fmt) 
11. fh.setFormatter(formatter) 
12.  
13. logger.addHandler(fh) 
14. return logger 
15.  
16. def doubler(number, logger): 
17. """ 
18. 可以被线程使用的一个函数 
19. """ 
20. logger.debug('doubler function executing') 
21. result = number * 2 
22. logger.debug('doubler function ended with: {}'.format( 
23. result)) 
24.  
25. if __name__ == '__main__': 
26. logger = get_logger 
27. thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina'] 
28. for i in range(5): 
29. my_thread = threading.Thread( 
30. target=doubler, name=thread_names[i], args=(i,logger)) 
31. my_thread.start 

代码中最大的改变就是加入了 get_logger函数。这段代码将创建一个被设置为调试级别的日志记录器。它将日志保存在当前目录(即脚本运行所在的目录)下，然后设置每行日志的格式。格式包括时间戳、线程名、日志记录级别以及日志信息。

在 doubler 函数中，我们把 print语句换成 logging 语句。你会注发现，在创建线程时，我们给 doubler 函数传入了 logger 对象。这样做的原因是，如果在每个线程中实例化 logging 对象，那么将会产生多个 logging 单例(singleton)，并且日志中将会有很多重复的内容。

最后，创建一个名称列表，然后使用 name关键字参数为每一个线程设置具体名称，这样就可以为线程命名。运行以上代码，将会得到包含以下内容的日志文件：

1. 2016-07-24 20:39:50,055 - Mike - DEBUG - doubler function executing 
2. 2016-07-24 20:39:50,055 - Mike - DEBUG - doubler function ended with: 0 
3. 2016-07-24 20:39:50,055 - George - DEBUG - doubler function executing 
4. 2016-07-24 20:39:50,056 - George - DEBUG - doubler function ended with: 2 
5. 2016-07-24 20:39:50,056 - Wanda - DEBUG - doubler function executing 
6. 2016-07-24 20:39:50,056 - Wanda - DEBUG - doubler function ended with: 4 
7. 2016-07-24 20:39:50,056 - Dingbat - DEBUG - doubler function executing 
8. 2016-07-24 20:39:50,057 - Dingbat - DEBUG - doubler function ended with: 6 
9. 2016-07-24 20:39:50,057 - Nina - DEBUG - doubler function executing 
10. 2016-07-24 20:39:50,057 - Nina - DEBUG - doubler function ended with: 8 

输出结果不言自明，所以继续介绍其他内容。在本节中再多说一点，即通过继承 threading.Thread实现多线程。举最后一个例子，通过继承 threading.Thread 创建子类，而不是直接调用 Thread 函数。

更新后的代码如下：

1. import logging 
2. import threading 
3.  
4. class MyThread(threading.Thread): 
5. def __init__(self, number, logger): 
6. threading.Thread.__init__(self) 
7. self.number = number 
8. self.logger = logger 
9.  
10. def run(self): 
11. """ 
12. 运行线程 
13. """ 
14. logger.debug('Calling doubler') 
15. doubler(self.number, self.logger) 
16.  
17. def get_logger: 
18. logger = logging.getLogger("threading_example") 
19. logger.setLevel(logging.DEBUG) 
20.  
21. fh = logging.FileHandler("threading_class.log") 
22. fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s' 
23. formatter = logging.Formatter(fmt) 
24. fh.setFormatter(formatter) 
25.  
26. logger.addHandler(fh) 
27. return logger 
28.  
29. def doubler(number, logger): 
30. """ 
31. 可以被线程使用的一个函数 
32. """ 
33. logger.debug('doubler function executing') 
34. result = number * 2 
35. logger.debug('doubler function ended with: {}'.format( 
36. result)) 
37.  
38. if __name__ == '__main__': 
39. logger = get_logger 
40. thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina'] 
41. for i in range(5): 
42. thread = MyThread(i, logger) 
43. thread.setName(thread_names[i]) 
44. thread.start 

这个例子中，我们只是创建一个继承于 threading.Thread的子类。像之前一样，传入一个需要翻一番的数字，以及 logging 对象。但是这次，设置线程名称的方式有点不太一样，变成了通过调用 thread 对象的setName方法来设置。不过仍然需要调用start来启动线程，不过你可能注意到我们并不需要在子类中定义该方法。当调用start时，它会通过调用run方法来启动线程。在我们的类中，我们调用 doubler 函数来做处理。输出结果中除了一些添加的额外信息内容几乎差不多。运行下这个脚本，看看你会得到什么。

线程锁与线程同步

当你有多个线程，就需要考虑怎样避免线程冲突。我的意思是说，你可能遇到多个线程同时访问同一资源的情况。如果不考虑这些问题并且制定相应的解决方案，那么在开发产品过程中，你总会在最糟糕的时候遇到这些棘手的问题。

解决办法就是使用线程锁。锁由 Python 的 threading 模块提供，并且它最多被一个线程所持有。当一个线程试图获取一个已经锁在资源上的锁时，该线程通常会暂停运行，直到这个锁被释放。来让我们看一个非常典型没有却应具备锁功能的例子：

1. import threading 
2.  
3. total = 0 
4.  
5. def update_total(amount): 
6. """ 
7. Updates the total by the given amount 
8. """ 
9. global total 
10. total += amount 
11. print (total) 
12. if __name__ == '__main__': 
13. for i in range(10): 
14. my_thread = threading.Thread( 
15. target=update_total, args=(5,)) 
16. my_thread.start 

如果往以上代码添加 time.sleep函数并给出不同长度的时间，可能会让这个例子更有意思。无论如何，这里的问题是，一个线程可能已经调用update_total函数并且还没有更新完成，此时另一个线程也有可能调用它并且尝试更新内容。根据操作执行顺序的不同，该值可能只被增加一次。

让我们给这个函数添加锁。有两种方法可以实现。第一种方式是使用 try/finally，从而确保锁肯定会被释放。下面是示例：

1. import threading 
2.  
3. total = 0 
4.  
5. lock = threading.Lock 
6. def update_total(amount): 
7. """ 
8. Updates the total by the given amount 
9. """ 
10. global total 
11. lock.acquire 
12. try: 
13. total += amount 
14. finally: 
15. lock.release 
16. print (total) 
17.  
18. if __name__ == '__main__': 
19. for i in range(10): 
20. my_thread = threading.Thread( 
21. target=update_total, args=(5,)) 
22. my_thread.start 

如上，在我们做任何处理之前就获取锁。然后尝试更新 total 的值，最后释放锁并打印出 total 的当前值。事实上，我们可以使用 Python 的 with语句避免使用 try/finally 这种较为繁琐的语句：

1. import threading 
2.  
3. total = 0 
4.  
5. lock = threading.Lock 
6.  
7. def update_total(amount): 
8. """ 
9. Updates the total by the given amount 
10. """ 
11. global total 
12. with lock: 
13. total += amount 
14. print (total) 
15.  
16. if __name__ == '__main__': 
17. for i in range(10): 
18. my_thread = threading.Thread( 
19. target=update_total, args=(5,)) 
20. my_thread.start 

正如你看到的那样，我们不再需要 try/finally作为上下文管理器，而是由with语句作为替代。

当然你也会遇到要在代码中通过多个线程访问多个函数的情况。当你第一次编写并发代码时，代码可能是这样的：

1. import threading 
2.  
3. total = 0 
4.  
5. lock = threading.Lock 
6. def do_something: 
7. lock.acquire 
8. try: 
9. print('Lock acquired in the do_something function') 
10. finally: 
11. lock.release 
12. print('Lock released in the do_something function') 
13. return "Done doing something" 
14.  
15. def do_something_else: 
16. lock.acquire 
17. try: 
18. print('Lock acquired in the do_something_else function') 
19. finally: 
20. lock.release 
21. print('Lock released in the do_something_else function') 
22. return "Finished something else" 
23.  
24. if __name__ == '__main__': 
25. result_one = do_something 
26. result_two = do_something_else 

这样的代码在上面的情况下能够正常工作，但假设你有多个线程都调用这两个函数呢。当一个线程正在运行这两个函数，然后另外一个线程也可能会修改这些数据，最后得到的就是不正确的结果。问题是，你甚至可能没有马上意识到结果错了。有什么解决办法呢?让我们试着找出答案。

通常首先想到的就是在调用这两个函数的地方上锁。让我们试着修改上面的例子，修改成如下所示：

1. import threading 
2.  
3. total = 0 
4.  
5. lock = threading.RLock 
6. def do_something: 
7.  
8. with lock: 
9. print('Lock acquired in the do_something function') 
10. print('Lock released in the do_something function') 
11. return "Done doing something" 
12.  
13.  
14. def do_something_else: 
15. with lock: 
16. print('Lock acquired in the do_something_else function') 
17. print('Lock released in the do_something_else function') 
18. return "Finished something else" 
19.  
20. def main: 
21. with lock: 
22. result_one = do_something 
23. result_two = do_something_else 
24. print (result_one) 
25. print (result_two) 
26.  
27. if __name__ == '__main__': 
28. main 

当你真正运行这段代码时，你会发现它只是挂起了。究其原因，是因为我们只告诉 threading 模块获取锁。所以当我们调用第一个函数时，它发现锁已经被获取，随后便把自己挂起了，直到锁被释放，然而这将永远不会发生。

真正的解决办法是使用重入锁(Re-Entrant Lock)。threading 模块提供的解决办法是使用RLock函数。即把lock = threading.lock替换为lock = threading.RLock，然后重新运行代码，现在代码就可以正常运行了。

如果你想在线程中运行以上代码，那么你可以用以下代码取代直接调用 main函数：

1. if __name__ == '__main__': 
2. for i in range(10): 
3. my_thread = threading.Thread( 
4. target=main) 
5. my_thread.start 

每个线程都会运行 main 函数，main 函数则会依次调用另外两个函数。最终也会产生 10 组结果集。

定时器

Threading 模块有一个优雅的 Timer类，你可以用它来实现在指定时间后要发生的动作。它们实际上会启动自己的自定义线程，通过调用常规线程上的start方法即可运行。你也可以调用它的cancel方法停止定时器。值得注意的是，你甚至可以在开始定时器之前取消它。

有一天，我遇到一个特殊的情况：我需要与已经启动的子进程通信，但是我需要它有超时处理。虽然处理这种特殊问题有很多不同的方法，不过我最喜欢的解决方案是使用 threading 模块的 Timer 类。

在下面这个例子中，我们将使用 ping指令作为演示。在 Linux 系统中，ping 命令会一直运行下去直到你手动杀死它。所以在 Linux 世界里，Timer 类就显得非常方便。示例如下：

1. import subprocess 
2. from threading import Timer 
3.  
4. kill = lambda process: process.kill 
5. cmd = ['ping', 'www.google.com'] 
6. ping = subprocess.Popen( 
7. cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 
8.  
9. my_timer = Timer(5, kill, [ping]) 
10. try: 
11. my_timer.start 
12. stdout, stderr = ping.communicate 
13. finally: 
14. my_timer.cancel 
15. print (str(stdout)) 

这里我们在 lambda 表达式中调用 kill 杀死进程。接下来启动 ping 命令，然后创建 Timer 对象。你会注意到，第一个参数就是需要等待的秒数，第二个参数是需要调用的函数，紧跟其后的参数是要调用函数的入参。在本例中，我们的函数是一个 lambda 表达式，传入的是一个只有一个元素的列表。如果你运行这段代码，它应该会运行 5 秒钟，然后打印出 ping 的结果。

其他线程组件

Threading 模块包含对其他功能的支持。例如，你可以创建信号量(Semaphore)，这是计算机科学中最古老的同步原语之一。基本上，一个信号量管理一个内置的计数器。当你调用acquire时计数器就会递减，相反当你调用release时就会递增。根据其设计，计数器的值无法小于零，所以如果正好在计数器为零时调用 acquire 方法，该方法将阻塞线程。

译者注：通常使用信号量时都会初始化一个大于零的值，如 semaphore = threading.Semaphore(2)

另一个非常有用的同步工具就是事件(Event)。它允许你使用信号(signal)实现线程通信。在下一节中我们将举一个使用事件的实例。

最后，在 Python 3.2 中加入了 Barrier对象。Barrier 是管理线程池中的同步原语，在线程池中多条线程需要相互等待对方。如果要传递 barrier，每一条线程都要调用wait方法，在其他线程调用该方法之前线程将会阻塞。全部调用之后将会同时释放所有线程。

线程通信

某些情况下，你会希望线程之间互相通信。就像先前提到的，你可以通过创建 Event对象达到这个目的。但更常用的方法是使用队列(Queue)。在我们的例子中，这两种方式都会有所涉及。下面让我们看看到底是什么样子的：

1. import threading 
2. from queue import Queue 
3.  
4. def creator(data, q): 
5. """ 
6. 生成用于消费的数据，等待消费者完成处理 
7. """ 
8. print('Creating data and putting it on the queue') 
9. for item in data: 
10. evt = threading.Event 
11. q.put((item, evt)) 
12.  
13. print('Waiting for data to be doubled') 
14. evt.wait 
15.  
16. def my_consumer(q): 
17. """ 
18. 消费部分数据，并做处理 
19.  
20. 这里所做的只是将输入翻一倍 
21.  
22. """ 
23. while True: 
24. data, evt = q.get 
25. print('data found to be processed: {}'.format(data)) 
26. processed = data * 2 
27. print(processed) 
28. evt.set 
29. q.task_done 
30.  
31. if __name__ == '__main__': 
32. q = Queue 
33. data = [5, 10, 13, -1] 
34. thread_one = threading.Thread(target=creator, args=(data, q)) 
35. thread_two = threading.Thread(target=my_consumer, args=(q,)) 
36. thread_one.start 
37. thread_two.start 
38.  
39. q.join 

让我们掰开揉碎分析一下。首先，我们有一个创建者(creator)函数(亦称作生产者(producer))，我们用它来创建想要操作(或者消费)的数据。然后用另外一个函数 my_consumer来处理刚才创建出来的数据。Creator 函数使用 Queue 的put方法向队列中插入数据，消费者将会持续不断的检测有没有更多的数据，当发现有数据时就会处理数据。Queue 对象处理所有的获取锁和释放锁的过程，这些不用我们太关心。

在这个例子中，先创建一个列表，然后创建两个线程，一个用作生产者，一个作为消费者。你会发现，我们给两个线程都传递了 Queue 对象，这两个线程隐藏了关于锁处理的细节。队列实现了数据从第一个线程到第二个线程的传递。当第一个线程把数据放入队列时，同时也传递一个 Event 事件，紧接着挂起自己，等待该事件结束。在消费者侧，也就是第二个线程，则做数据处理工作。当完成数据处理后就会调用 Event 事件的 set方法，通知第一个线程已经把数据处理完毕了，可以继续生产了。

最后一行代码调用了 Queue 对象的 join方法，它会告知 Queue 等待所有线程结束。当第一个线程把所有数据都放到队列中，它也就运行结束了。

结束语