8.10 并发执行

多线程类似于同时执行多个不同程序，多线程运行有如下优点：

• 使用线程可以把占据长时间的程序中的任务放到后台去处理。
• 用户界面可以更加吸引人，这样比如用户点击了一个按钮去触发某些事件的处理，可以弹出一个进度条来显示处理的进度
• 程序的运行速度可能加快
• 在一些等待的任务实现上如用户输入、文件读写和网络收发数据等，线程就比较有用了。在这种情况下我们可以释放一些珍贵的资源如内存占用等等。

8.10.1 threading 线程模块

threading模块提供了Thread类来处理线程，Thread类提供了以下方法:

• run(): 用以表示线程活动的方法。
• start(): 启动线程活动。
• join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。
• isAlive(): 返回线程是否活动的。
• getName(): 返回线程名。
• setName(): 设置线程名。

使用 threading 模块创建线程

我们可以通过直接使用threading.Thread来创建一个新的线程，传递的参数分别为线程的执行函数以及线程名。

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
调用这个构造函数时，必需带有关键字参数。参数如下：
group 应该为 None；为了日后扩展 ThreadGroup 类实现而保留。
target 是用于 run() 方法调用的可调用对象。默认是 None，表示不需要调用任何方法。
name 是线程名称。默认情况下，由 "Thread-N" 格式构成一个唯一的名称，其中 N 是小的十进制数。
args 是用于调用目标函数的参数元组。默认是 ()。
kwargs 是用于调用目标函数的关键字参数字典。默认是 {}。

例如：8.9-threading.py

import threading
import time


def run():
    print("开始线程：" + threading.current_thread().name)
    for i in range(5):
        print('线程 {} >>> {}'.format(threading.current_thread().name, i + 1))
        time.sleep(1)
    print("退出线程：" + threading.current_thread().name)


if __name__ == '__main__':
    # 创建新线程
    thread1 = threading.Thread(target=run, name='run_thread_1')
    thread2 = threading.Thread(target=run, name='run_thread_2')

    # 开启新线程
    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()
    print("退出主线程")

运行结果为：

开始线程：run_thread_1
线程 run_thread_1 >>> 1
开始线程：run_thread_2
线程 run_thread_2 >>> 1
线程 run_thread_1 >>> 2
线程 run_thread_2 >>> 2
线程 run_thread_1 >>> 3
线程 run_thread_2 >>> 3
线程 run_thread_1 >>> 4
线程 run_thread_2 >>> 4
线程 run_thread_1 >>> 5
线程 run_thread_2 >>> 5
退出线程：run_thread_1
退出线程：run_thread_2
退出主线程

8.10.2 线程同步

多线程的优势在于可以同时运行多个任务（至少感觉起来是这样）。但是当线程需要共享数据时，可能存在数据不同步的问题：多个线程共同对某个数据修改，则可能出现不可预料的结果。为了保证数据的正确性，需要对多个线程进行同步。

例如：8-10-threading-withoutlock.py

import threading

g_ticket = 20


def run():
    global g_ticket
    print("开启线程： " + threading.current_thread().name)
    while g_ticket > 0:
        print('{} 卖出第 {} 张票。'.format(threading.current_thread().name, (20 - g_ticket + 1)))
        g_ticket -= 1


if __name__ == '__main__':
    # 创建新线程
    thread1 = threading.Thread(target=run, name='run_thread_1')
    thread2 = threading.Thread(target=run, name='run_thread_2')

    # 开启新线程
    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()
    print("退出主线程")

结果中的第二张票被卖出两次，没有实现正常的卖票效果。出现这种情况的原因是因为两个线程在并行操作同一个对象，所以会出现两个线程都卖出了同一张票的情况，但是这显然是不对的。

开启线程： run_thread_1
run_thread_1 卖出第 1 张票。
开启线程： run_thread_2
run_thread_2 卖出第 2 张票。
run_thread_1 卖出第 2 张票。
run_thread_2 卖出第 3 张票。
run_thread_1 卖出第 4 张票。
run_thread_2 卖出第 5 张票。
run_thread_1 卖出第 6 张票。
run_thread_2 卖出第 7 张票。
run_thread_1 卖出第 8 张票。
run_thread_2 卖出第 9 张票。
run_thread_1 卖出第 10 张票。
run_thread_2 卖出第 11 张票。
run_thread_1 卖出第 12 张票。
run_thread_2 卖出第 13 张票。
run_thread_1 卖出第 14 张票。
run_thread_2 卖出第 15 张票。
run_thread_1 卖出第 16 张票。
run_thread_2 卖出第 17 张票。
run_thread_2 卖出第 19 张票。
run_thread_2 卖出第 20 张票。
run_thread_1 卖出第 18 张票。
退出主线程

使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间。

例如：8.11-threading-withlock.py

import threading
import time

g_ticket = 20
threadLock = threading.Lock()


def run():
    global g_ticket
    print("开启线程： " + threading.current_thread().name)
    # 获取锁，用于线程同步
    while g_ticket > 0:
        threadLock.acquire()
        if g_ticket > 0:
            g_ticket -= 1
        else:
            break
        print('{} 卖出第 {} 张票。'.format(threading.current_thread().name, (20 - g_ticket)))
        time.sleep(1)
        threadLock.release()


if __name__ == '__main__':
    # 创建新线程
    thread1 = threading.Thread(target=run, name='run_thread_1')
    thread2 = threading.Thread(target=run, name='run_thread_2')

    # 开启新线程
    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()
    print("退出主线程")

运行结果为：（结果可能不同）

开启线程： run_thread_1
run_thread_1 卖出第 1 张票。
开启线程： run_thread_2
run_thread_1 卖出第 2 张票。
run_thread_1 卖出第 3 张票。
run_thread_1 卖出第 4 张票。
run_thread_1 卖出第 5 张票。
run_thread_2 卖出第 6 张票。
run_thread_2 卖出第 7 张票。
run_thread_2 卖出第 8 张票。
run_thread_2 卖出第 9 张票。
run_thread_1 卖出第 10 张票。
run_thread_2 卖出第 11 张票。
run_thread_1 卖出第 12 张票。
run_thread_2 卖出第 13 张票。
run_thread_2 卖出第 14 张票。
run_thread_2 卖出第 15 张票。
run_thread_2 卖出第 16 张票。
run_thread_2 卖出第 17 张票。
run_thread_2 卖出第 18 张票。
run_thread_2 卖出第 19 张票。
run_thread_1 卖出第 20 张票。
退出主线程

8.10.3 GIL——Python多线程并不能真正能发挥作用

GIL 全称 Global Interpreter Lock，官方的解释如下：

In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)

GIL 是 CPython 的特性，由于 CPython 是大部分环境下默认的 Python 执行环境，所以多数情况下要考虑 GIL 的影响。但 GIL 并不是 Python 的特性，Python 完全可以不依赖于 GIL。

GIL 保证在同一个时间只能有一个线程执行任务，防止多线程并发执行机器码，也就是多线程并不是真正的并发，只是交替的执行。

Python 社区目前在不断改进 GIL，甚至是尝试去除 GIL，并在各个版本中有了不少的进步。

实际开发中需根据任务类型选择合适的并发方案：

• I/O 密集型任务：多线程仍为有效选择。
• CPU 密集型任务：优先考虑多进程或利用 C 扩展释放 GIL 的库。

例如，对于一个循环一定次数的计数器函数，一种情况是单线程执行若干次，另一个是多线程执行，最后比较总的执行时间。

8.12-single_thread.py

import threading
import time


def my_counter():
    i = 0
    for _ in range(100000000):
        i = i + 1
    return True


def main():
    start_time = time.time()
    for tid in range(3):
        t = threading.Thread(target=my_counter)
        t.start()
        t.join()
    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))


if __name__ == '__main__':
    main()

8.13-multi_thread.py

import threading
import time


def my_counter():
    i = 0
    for _ in range(100000000):
        i = i + 1
    return True


def main():
    thread_array = {}
    start_time = time.time()

    for tid in range(3):
        t = threading.Thread(target=my_counter)
        t.start()
        thread_array[tid] = t
    for i in range(3):
        thread_array[i].join()
    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))


if __name__ == '__main__':
    main()

在双核CPU环境下，不同的python版本表现结果如下：

python 2.7:

$ python -V
Python 2.7.15rc1

$ python 8.12-single_thread.py
Total time: 17.3186781406

$ python 8.13-multi_thread.py
Total time: 38.7809410095

python 3.7:

>python -V
Python 3.7.4

>python 8.12-single_thread.py
Total time: 15.218140363693237

>python 8.13-multi_thread.py
Total time: 15.502634763717651

可以看到在python2系列中，多线程的工作方式甚至会比单线程更慢，而在python3系列中，多线程的工作方式也仅仅是使用了和单线程相同的时间，多核的优势并未体现出来。

8.10.4 多进程

Python可以使用多进程的方式代替多线程，并提供了一个跨平台的多进程支持。multiprocessing模块就是跨平台版本的多进程模块。multiprocessing模块提供了一个Process类来代表一个进程对象。

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

group 应该始终是 None ；它仅用于兼容 threading.Thread 。 
target 是由 run() 方法调用的可调用对象。它默认为 None ，意味着什么都没有被调用。 
name 是进程名称。 
args 是目标调用的参数元组。 
kwargs 是目标调用的关键字参数字典。

其余方法等同于线程。

下面的例子演示了启动一个子进程并等待其结束：8.14-multiprocess.py

import multiprocessing
import os


def run_proc(name):
    print('子进程名：', multiprocessing.current_process().name)
    print('运行子进程，参数：{}， 进程ID：{}...'.format(name, os.getpid()))


if __name__ == '__main__':
    print('父进程ID：{}.'.format(os.getpid()))
    p = multiprocessing.Process(target=run_proc, name='p1', args=('abc',))
    print('准备启动子进程.')
    p.start()
    p.join()
    print('子进程结束.')
    print('父进程结束')

执行结果为：

父进程ID：19772.
准备启动子进程.
子进程名： p1
运行子进程，参数：abc， 进程ID：22652...
子进程结束.
父进程结束

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。

8.10.5 进程间通信

Process之间需要通信，操作系统提供了很多机制来实现进程间的通信。Python的multiprocessing模块包装了底层的机制，提供了Queue、Pipes等多种方式来交换数据。

Python 的 Queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出）队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列 PriorityQueue。

这些队列都实现了锁原语（可以理解为原⼦操作，即要么不做，要么就做完），能够在多线程中直接使用，可以使用队列来实现线程间的同步。

Queue 模块中的常用方法:

• Queue.qsize() 返回队列的大小
• Queue.empty() 如果队列为空，返回True,反之False
• Queue.full() 如果队列满了，返回True,反之False
• Queue.full 与 maxsize 大小对应
• Queue.get([block[, timeout]])获取队列，timeout等待时间
• Queue.get_nowait() 相当Queue.get(False)
• Queue.put(item) 写入队列，timeout等待时间
• Queue.put_nowait(item) 相当Queue.put(item, False)
• Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
• Queue.join() 实际上意味着等到队列为空，再执行别的操作

我们以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：8.15-queue.py

import multiprocessing
import os
import time
import random


def write(q):
    print('写入进程ID：{}'.format(os.getpid()))
    for value in ['A', 'B', 'C']:
        print('写入 {} 到queue中...'.format(value))
        q.put(value)
        time.sleep(random.random())


def read(q):
    print('读取进程ID：{}'.format(os.getpid()))
    while True:
        value = q.get(True)
        print('从queue中读取：{}'.format(value))


if __name__ == '__main__':
    # 父进程创建Queue，并传给各个子进程：
    q = multiprocessing.Queue()
    pw = multiprocessing.Process(target=write, args=(q,))
    pr = multiprocessing.Process(target=read, args=(q,))
    # 启动子进程pw，写入:
    pw.start()
    # 启动子进程pr，读取:
    pr.start()
    # 等待pw结束:
    pw.join()
    # pr进程里是死循环，无法等待其结束，只能强行终止:
    pr.terminate()

运行结果如下：

写入进程ID：2984
写入 A 到queue中...
读取进程ID：7964
从queue中读取：A
写入 B 到queue中...
从queue中读取：B
写入 C 到queue中...
从queue中读取：C

利用队列实现上面的多进程卖票：8.16-多进程队列卖票.py

import multiprocessing
import time
import random


def sell(q):
    name = multiprocessing.current_process().name
    print('卖票进程：', name)
    while not q.empty():
        value = q.get(True)
        print('进程{}卖了第{}张票。'.format(name, value))
        time.sleep(random.random())


if __name__ == '__main__':
    # 父进程创建Queue，并传给各个子进程：
    q = multiprocessing.Queue()
    for x in range(1, 21):
        q.put(x)

    p1 = multiprocessing.Process(target=sell, name='A', args=(q,))
    p2 = multiprocessing.Process(target=sell, name='B', args=(q,))
    p3 = multiprocessing.Process(target=sell, name='C', args=(q,))
    # 启动子进程:
    p1.start()
    p2.start()
    p3.start()

    # 等待结束:
    p1.join()
    p2.join()
    p3.join()

结果为：

卖票进程： B
进程B卖了第1张票。
卖票进程： A
进程A卖了第2张票。
卖票进程： C
进程C卖了第3张票。
进程A卖了第4张票。
进程A卖了第5张票。
进程C卖了第6张票。
进程B卖了第7张票。
进程B卖了第8张票。
进程B卖了第9张票。
进程A卖了第10张票。
进程C卖了第11张票。
进程C卖了第12张票。
进程A卖了第13张票。
进程C卖了第14张票。
进程B卖了第15张票。
进程B卖了第16张票。
进程B卖了第17张票。
进程A卖了第18张票。
进程C卖了第19张票。
进程C卖了第20张票。

8.10.6 多进程对多核CPU的利用

使用多进程的方法，改写前面8.13-multi_thread.py：

8.17-multi_process.py

import multiprocessing
import time


def my_counter():
    i = 0
    for _ in range(100000000):
        i = i + 1
    return True


def main():
    process_array = {}
    start_time = time.time()

    for pid in range(3):
        p = multiprocessing.Process(target=my_counter)
        p.start()
        process_array[pid] = p
    for i in range(3):
        process_array[i].join()
    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))


if __name__ == '__main__':
    main()

运行结果如下：

$ python -V
Python 2.7.15rc1

$ python 8.17-multi_process.py
Total time: 8.47458720207

>python -V
Python 3.7.4

>python 8.17-multi_process.py
Total time: 8.504215717315674

可以看出，使用多进程方式后，python2 与 python3 的结果相差不大，并且都比单线程执行的时间要短，能够充分利用多核CPU的资源。