Python 并发编程

Python 中的线程会在一个单独的系统级线程（关于系统级（或称内核级）线程与用户级线程，可以再进一步了解）中执行（比如说一个 POSIX 线程或者一个 Windows 线程），这些线程将由操作系统来全权管理。

from threading import Thread
t = Thread(target=a_function, args=(10,))
t.start()

你可以通过 t.is_alive() 来查看一个线程是否在执行，也可以通过 t.join() 将其加入到当前线程，并等待其执行完成。

除此之外，你无法对线程多更多的控制。线程一旦启动，就将独立执行直到目标函数结束。

如果你确实想向线程发送信号来控制线程，你需要自己实现它。比如说你修改某个值，然后让线程轮询这个值。

普通线程与后台线程（又称守护线程）
#

当 Python 主线程代码执行完后，主线程是否退出，取决于当前是否有非后台线程仍在运行，有的话，主线程会等待它们运行结束后再退出，否则（没有线程或只有后台线程仍在运行），主线程将立即退出。主线程退出，所有的后台线程也就结束。

要明确的是，GIL 只会影响到那些 CPU 任务重的程序。如果你的程序大部分只是因为涉及到 I/O、网络交互而需要用到多线程，那么你大可放心，即使创建几千个 Python 线程，现代计算机也不会有什么压力。

许多程序员在遭遇线程性能问题的时候，立马就怪罪 GIL，往往是不厚道或太天真。

GIL 限制
#

由于全局解释锁（GIL）的限制，Python 中的线程并不是真正同时运行的线程，解释器实际上同一时刻只执行其中一个线程。这一限制是的 Python 的多线程并不能利用多核 CPU 的优势。所以 Python 的多线程更适合于处理 I/O 操作或其他一些需要并发执行的阻塞操作，而并不适合于计算密集型任务的并发处理。

Event 事件
#

由于 Python 中的线程是独立运行的，我们无法向其发送信号，也不知道其运行得怎么样，运行到了哪一步，而往往线程之间又是需要协作的，比如说我们可能在 A 线程执行到某处时，想要等待其他线程完成了某件事情之后，A 再往下执行。如果使用共享变量的话，事情会变得复杂，而 event 事件则能优雅地应对此类场景。

from threading import Thread, Event

evt = Event()

# A 线程的目标方法
def thread_func_A(evt):
	# do something
	evt.wait()
	# do something

# B 线程的目标方法
def thread_func_B(evt):
	# do something
	evt.set()
	# do something

在如上的例子中，线程 A 的目标方法会在 evt.wait() 等待，待线程 B 的目标方法执行了 evt.set() 之后，线程 A 才会继续往下执行。

尽管 event 还提供了 clear() 方法对其进行重制，但是一个 event 最好作为一次性使用。因为 event 的 clear 操作容易带来逻辑冲突以及错过处理，甚至死锁等问题。

如果你想要一个可以重复使用的 event，那么你可以使用 Condition 对象。另外，event 还有一个特点是，当其被 set 的时候，所有正在等待它的线程都会被唤醒。如果你指向唤醒单个线程，也应该用 Condition。

Condition 提供了以下方法：

wait()

等待其他线程发送通知（notification）之后再往下执行。
notify(n=1)

发送通知，允许 n 个（如果有的话）等待此通知的线程被唤醒。
notify_all()

发送通知，允许所有等待此通知的线程被唤醒。

使用 Queue 进行线程间通信
#

编写涉及到大量线程同步问题的代码会让你痛不欲生，比较建议的方式是使用队列来进行线程间通信，或者把每个线程当作一个 Actor，利用 Actor 模型来控制并发。此处我们讲解线程间通信。

一个线程向其他线程发送数据，最安全的方式是使用 queue 库中的队列。创建一个被多个线程共享的 Queue 对象，这些线程通过使用 put() 和 get() 操作来向队列中添加或者删除元素。

Queue 对象已经包含了必要的锁，所以可以通过它在多个线程间安全地共享数据。

（Queue 不是一个普通的列表，应当对其详细了解。）

在使用队列时，协调生产者和消费者的关闭问题是一个麻烦，通常的方案是在队列中放一个特殊值，当消费者读到这个值时，终止执行。另外，如果存在多个消费者的话，某个消费者通过 get() 取到特殊值后，应该再将其 put() 回去，这样，其他的消费者也能够取到这个值并结束。

使用队列进行进程间通信是一个单向、不确定的过程。如果生产者想要知道消费者是否接收并处理好了数据，则生产者可以将一个 event 对象和数据一起发送，Queue 和 event 相结合。

from queue import Queue
from threading import Thread, Event

# A thread that produces data
def producer(out_q):
    while running:
        # Produce some data
        ...
        # Make an (data, event) pair and hand it to the consumer
        evt = Event()
        out_q.put((data, evt))
        ...
        # Wait for the consumer to process the item
        evt.wait()

# A thread that consumes data
def consumer(in_q):
    while True:
        # Get some data
        data, evt = in_q.get()
        # Process the data
        ...
        # Indicate completion
        evt.set()

我们还可以将 Queue 和 event、Condition 进行更丰富的封装，以实现更复杂的通信机制。

基于队列编写多线程在多数情况下都是比较明智的选择，由于其本身实现了安全的同步机制，为你免去了许多烦恼。此外，使用 Queue 这种基于消息队列的通信机制，也方便你将代码移植到基于消息队列通信的分布式系统。

带锁的共享变量
#

为避免多线程的竞态冲突，我们要对共享变量加锁，以保证原子性操作。

import threading

class SharedCounter:
    '''
    A counter object that can be shared by multiple threads.
    '''
    def __init__(self, initial_value = 0):
        self._value = initial_value
        self._value_lock = threading.Lock()

    def incr(self,delta=1):
        '''
        Increment the counter with locking
        '''
        with self._value_lock:
             self._value += delta

    def decr(self,delta=1):
        '''
        Decrement the counter with locking
        '''
        with self._value_lock:
             self._value -= delta

Lock 对象应该和 with 语句一起使用，以保证互斥执行，即每次只有一个线程可以执行 with 语句包含的代码块。with 语句会在这个代码块执行前自动获取锁，在执行结束后自动释放锁。

相较于显式地调用 lock.acquire() 和 lock.release()，with 语句更加优雅，也不易出错，尤其是可以避免程序员忘了进行 release() 。

为了避免出现死锁，我们应到保证一个线程一次只允许获取一个锁，不许嵌套。不然是很容易发生 A 在等 B 释放，B 在等 A 释放的死锁情况的。

如果你真的必须要在同一个线程中获取多个锁，有一个避免死锁的方案就是为每一个锁分配一个唯一的 id，然后只允许按照升序规则来使用多个锁，不是说得逐一获取，只要不存在逆序就行。

作用域为线程的变量
#

如果你想要这样一个变量，它在当前线程内就像是一个全局变量，当前线程内的所有函数都可以使用这个变量，而对其它线程又是不可见的。那么 threading.loca() 就是你想要的。

import threading
    
# 创建全局ThreadLocal对象:
local_school = threading.local()

def student_greet():
    # 获取当前线程关联的student:
    student = local_school.student
    print('Hello, %s (in %s)' % (student, threading.current_thread().name))

def student_thread(student):
    # 绑定ThreadLocal的student:
    local_school.student = student
    student_greet()

t1 = threading.Thread(target= student_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= student_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

这个实现原理其实很简单，在如上代码中，local_school 本身是一个全局变量，它实际上是为每一个线程维护着一个单独的实例字典，所以当我们在某一个线程中访问 local_school.student 的时候，访问的其实是 local_school 中针对你当前这个线程实例所映射的字典中存储的那个 student。

线程池
#

线程池是什么呢，它是我们预先创建出来的一揽子线程。它的价值在于，当我们有几个任务需要线程去做时，我们可以把这几个任务给线程池中的空闲线程，让它们去做。而线程池中的线程执行完任务后，它不会被系统回收，而是会回到空闲状态，等待你再次给它分配任务。

这样有什么好处呢？我们知道，Cpython 的线程是内核线程，是由操作系统来创建和管理的。所以一个线程的创建、销毁其实是挺消耗资源和机能的。所以如果我们创建出线程池，并且可以反复地利用它们，免去反复的创建和销毁，这也是一种性能优势。

此外，线程池也方便我们对线程并发数量进行控制。

from concurrent.futures import ThreadPoolExecutor
# 创建线程池
thread_pool = ThreadPoolExecutor(5)
# 向线程池中提交任务
future_task = pool.sbumit(task_function, args)

我们通过调用 future_task.done() 可以得知任务是否执行完成，通过调用 future_task.result() 可以得到任务的返回值。

普通线程与后台线程（又称守护线程） #

GIL 限制 #

Event 事件 #

使用 Queue 进行线程间通信 #

带锁的共享变量 #

作用域为线程的变量 #

线程池 #