多线程

如果你是 Python 的使用者，这些话你肯定听到过无数次：Python 的多线程是假的，Python 的多线程是鸡肋，Python 的多线程没有意义。说这些话的人其中有挺大一部分，也无非就是知道一个 GIL，甚至 GIL 到底 lock 的是什么，可能也不那么清楚。Python 的多线程确实有它的问题，但是说多线程没有意义，显然就言过其实了。今天我们就来聊聊 Python 的多线程，能干啥？

多线程的用途

很多人可能觉得多线程唯一的价值就是利用多核 CPU 提升程序的运行速度。但是实际上，第一个多核 CPU，IBM 的 POWER4，出现在 2001 年。而在个人电脑上，我们更熟悉的英特尔、AMD，是在 2005 年才推出的它们的第一款多核 CPU。可是，多线程这个概念，在上个世纪五六十年代就已经出现了。那这中间接近半个世纪的时间里，多线程都是没有意义的吗？

多线程出现之初啊，解决的核心问题是，在不显式地切换任务的情况下，让 CPU 可以同时处理若干个任务。而这个价值，直至今日，依然存在。这里主要有两件事，值得一提。

利用 I/O 等待

第一个是我们相对讨论得更多的，叫做 I/O 瓶颈任务。什么意思呢？就是在单线程的情况下，我们正常写程序，对吧？在遇到比较慢的 I/O 任务，比如说这个文件读取啊、网络传输之类的事情的时候，是需要原地干等着的，对吧？要等这个 I/O 任务完成了，才能接着做下面的事情。那在这段时间，CPU 实际上就在闲着，是吧？那通过多线程呢，我们可以把这段闲着的时间给利用起来，干点别的。

我们给大家举个例子啊。大家看这边我写了一个非常简单的爬虫程序，它的核心代码就在这个 run 函数里面，对于你给它的每一个单词，下载它的 wiki page。这里我们简单地做了两个方案。第一个方案是把所有的单词用一个线程来下载，第二个方案是把这些单词平均分成两份，然后用两个线程来下载。下面我们就做了一个非常简陋的计时程序，算一算需要多长时间。

import threading
import time

import requests

words = list(range(6))


class Task(threading.Thread):
    def __init__(self, words):
        self.words = words
        self.total_word = 0
        super().__init__()

    def run(self):
        for word in self.words:
            requests.get(f"https://en.wikipedia.org/wiki/{word}")


t = Task(words)
t1 = Task(words[: len(words) // 2])
t2 = Task(words[len(words) // 2:])

start = time.time()
t.start()
t.join()

print(time.time() - start)

start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()

print(time.time() - start)

多运行几次我们可以看到，单线程的时候差不多是三秒多一点，而两个线程大概是一秒八左右。算下来差不多两个线程比一个线程快了一倍。这个就是在 I/O 瓶颈下，多线程对程序的提速效果。

协程的放权

当然对于 I/O 瓶颈的任务，我们现在经常可以使用协程，也就是 asyncio 来完成。协程相对线程更轻，而且由于是单线程，没有竞争冒险的问题，所以在网络传输的应用里面使用的比例越来越高了。

那协程的出现就宣判了多线程的死刑吗？也不是啊。尽管协程有着很多优质的特性，但是它也有一个无法回避的问题，就是必须要显式地放权。在 Python 中就是必须要通过 await 的形式，去把控制权交还给 event loop。

在有些情况下，比如你所有的程序本质上都是大量的 I/O 操作的时候，这不是个问题，因为每一个协程的 task 都会频繁地放权。

但是有些情况下，单纯的协程就不太好用。比如你有一个大量计算的任务，但是不着急，可以慢慢做，又有很多小的任务需要在低延迟下解决。这个时候怎么处理协程的调度，就是一个麻烦的事情。

人工放权

如果你不做任何处理，你的协程就会陷入到计算任务中出不来，没法及时地处理那些需要低延迟的小任务。可如果你在计算任务中人工安置一些放权点，又会让你这个任务看起来很奇怪。

下面这段程序就是这么一个场景：long_task 是那个慢慢做也没关系的计算任务，我们用递归的 fib(25) 来代替；short_task 是需要低延迟的小任务，每 10 毫秒要被叫醒一次，一共有四个。为了能在计算的中途放权，fib 被写成了 coroutine function，每一层递归都得 await 一次，注释掉的那行就是人工安置的放权点。

import asyncio


async def long_task():
    async def fib(n):
        if n <= 1:
            return 1
        # await asyncio.sleep(0)
        return await fib(n - 1) + \
            await fib(n - 2)

    while True:
        await fib(25)
        await asyncio.sleep(0)


async def short_task():
    while True:
        await asyncio.sleep(0.01)


async def main():
    task_list = [short_task() for _ in range(4)]
    task_list.append(long_task())
    await asyncio.gather(*task_list)


asyncio.run(main())

更何况这还需要你对这个任务的消耗时间足够了解。

第三方函数无法放权

不仅如此，如果你调用的是别人的函数，你是没法在别人的函数里面放权的。

import asyncio


def fib(n):
    if n <= 1:
        return 1
    return fib(n - 1) + fib(n - 2)


async def long_task():
    while True:
        # imported from library
        fib(25)
        await asyncio.sleep(0)


async def short_task():
    while True:
        await asyncio.sleep(0.01)


async def main():
    task_list = [short_task() for _ in range(4)]
    task_list.append(long_task())
    await asyncio.gather(*task_list)


asyncio.run(main())

long_task 每进去一次，事件循环就得等它整个算完才能回来。把这段程序跑起来，记下 short_task 每次被唤醒的时刻，前 600 毫秒是这样的：

long_task 在事件循环里算 fib(25) 时，short_task 的唤醒间隔

最上面那条是 long_task，每个方块是一次 fib(25) 的计算；下面四条是四个 short_task，每根竖线代表它被唤醒了一次。

线程的抢占式切换

那多线程的存在就可以比较好地解决这种问题。由于操作系统的底层机制会自动帮你切换线程，你就不需要担心一个任务把 CPU 占死，然后其他人喝不到汤的情况。

import asyncio
import threading


def fib(n):
    if n <= 1:
        return 1
    return fib(n - 1) + fib(n - 2)

def long_task():
    while True:
        fib(25)

async def short_task():
    while True:
        await asyncio.sleep(0.01)

async def main():
    task_list = [short_task() for _ in range(4)]
    t = threading.Thread(target=long_task)
    t.start()
    await asyncio.gather(*task_list)

asyncio.run(main())

fib 还是那个没法放权的普通函数，只是 long_task 被丢进了一条自己的线程。同样跑一遍，short_task 密了一个数量级：

long_task 放进单独的线程里算 fib(25) 时，short_task 的唤醒间隔

run_in_executor

当然了，你现在可以在 asyncio 里面用这个 run_in_executor 来解决这个问题。但是我们要知道这个解决方案的背后，它的本质它就是多线程，当然也有可能是多进程。

import asyncio


def fib(n):
    if n <= 1:
        return 1
    return fib(n - 1) + fib(n - 2)


async def long_task():
    while True:
        await asyncio.get_event_loop().run_in_executor(None, fib, 25)
        await asyncio.sleep(0)


async def short_task():
    while True:
        await asyncio.sleep(0.01)


async def main():
    task_list = [short_task() for _ in range(4)]
    task_list.append(long_task())
    await asyncio.gather(*task_list)


asyncio.run(main())

说了这么多就是想告诉大家，尽管由于 GIL 的存在，Python 没有办法像其他的一些语言一样，通过多线程来利用多核，提升计算瓶颈的程序运行速度，但是 Python 当中的多线程也完全是有其存在价值的。

你需要去了解它能做什么，不能做什么，有什么优势，有什么劣势，然后在不同的场景下使用合适的工具，而不是人云亦云，抱着网上听来的只言片语，奉为圭臬。

多线程的用途​

利用 I/O 等待​

协程的放权​

人工放权​

第三方函数无法放权​

线程的抢占式切换​

run_in_executor​