python - 这个多线程函数是异步的吗

Question

恐怕我仍然有点困惑(尽管检查了其他线程)是否:

所有异步代码都是多线程的

所有多线程函数都是异步的

我最初的猜测是否定的，正确的异步代码应该能够在一个线程中运行 - 但是可以通过添加线程来改进它，例如:



所以我构建了这个玩具示例:

from threading import *
from queue import Queue
import time

def do_something_with_io_lag(in_work):
    out = in_work
    # Imagine we do some work that involves sending
    # something over the internet and processing the output
    # once it arrives
    time.sleep(0.5) # simulate IO lag
    print("Hello, bee number: ",
          str(current_thread().name).replace("Thread-",""))

class WorkerBee(Thread):
    def __init__(self, q):
        Thread.__init__(self)
        self.q = q

    def run(self):
        while True:
            # Get some work from the queue
            work_todo = self.q.get()
            # This function will simiulate I/O lag
            do_something_with_io_lag(work_todo)
            # Remove task from the queue
            self.q.task_done()

if __name__ == '__main__':
    def time_me(nmbr):
        number_of_worker_bees = nmbr
        worktodo = ['some input for work'] * 50

        # Create a queue
        q = Queue()
        # Fill with work
        [q.put(onework) for onework in worktodo]
        # Launch processes
        for _ in range(number_of_worker_bees):
            t = WorkerBee(q)
            t.start()
        # Block until queue is empty
        q.join()

    # Run this code in serial mode (just one worker)
    %time time_me(nmbr=1)
    # Wall time: 25 s
    # Basically 50 requests * 0.5 seconds IO lag
    # For me everything gets processed by bee number: 59

    # Run this code using multi-tasking (launch 50 workers)
    %time time_me(nmbr=50)
    # Wall time: 507 ms
    # Basically the 0.5 second IO lag + 0.07 seconds it took to launch them
    # Now everything gets processed by different bees

它是异步的吗？

对我来说，这段代码似乎不是异步的，因为它是我的示例图中的图 3。 I/O 调用阻塞了线程(虽然我们感觉不到，因为它们是并行阻塞的)。

但是，如果是这种情况，我很困惑为什么 requests-futures 被认为是异步的，因为它是 ThreadPoolExecutor 的包装器:

with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
    future_to_url = {executor.submit(load_url, url, 10): url for url in     get_urls()}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()

这个功能只能在一个线程上运行吗？

特别是与 asyncio 相比时，这意味着它可以运行单线程

There are only two ways to have a program on a single processor do “more than one thing at a time.” Multi-threaded programming is the simplest and most popular way to do it, but there is another very different technique, that lets you have nearly all the advantages of multi-threading, without actually using multiple threads. It’s really only practical if your program is largely I/O bound. If your program is processor bound, then pre-emptive scheduled threads are probably what you really need. Network servers are rarely processor bound, however.

Answer 1

首先，一个说明: concurrent.futures.Future 与 asyncio.Future 不一样.基本上它只是一个抽象 - 一个对象，它允许您在分配作业之后但在完成之前引用程序中的作业结果(或异常，也是结果)。这类似于将普通函数的结果分配给某个变量。

多线程 :关于您的示例，当使用多个线程时，您可以说您的代码是“异步”的，因为多个操作同时在不同线程中执行，而无需等待彼此完成，您可以在计时结果中看到它。你是对的，你的功能归功于 sleep正在阻塞，它会在指定的时间内阻塞工作线程，但是当您使用多个线程时，这些线程会被并行阻塞。所以，如果你想找一份工作 sleep另一个没有并运行多个线程，一个没有 sleep将执行计算，而另一个将 sleep 。当您使用单线程时，作业以串行方式一个接一个地执行，因此当一个作业休眠时，其他作业等待它，实际上直到轮到它们时它们才存在。您的时间测试几乎证明了所有这些。事情发生在 print与“线程安全”有关，即打印使用标准输出，这是一个单一的共享资源。因此，当您的多个线程尝试同时打印时，切换发生在内部，并且您得到了奇怪的输出。 (这也显示了多线程示例的“异步性”。)为了防止此类错误，有锁定机制，例如锁、信号量等。

异步 :为了更好地理解目的，请注意“IO”部分，它不是“异步计算”，而是“异步输入/输出”。在谈论 asyncio 时，您通常一开始不会考虑线程。 Asyncio 是关于事件循环和生成器(协程)的。事件循环是仲裁器，它管理注册到循环的协程(及其回调)的执行。协程被实现为生成器，即允许迭代执行某些操作的函数，在每次迭代和“返回”时保存状态，并在下一次调用时继续保存的状态。所以基本上事件循环是 while True:循环，它一个接一个地调用分配给它的所有协程/生成器，并且它们在每个这样的调用中提供结果或无结果 - 这为“异步”提供了可能性。 (一种简化，因为有调度机制，可以优化这种行为。)这种情况下的事件循环可以在单线程中运行，如果协程是非阻塞的，它会给你真正的“异步性”，但如果它们是阻塞的，那么它基本上是线性执行。

您可以使用显式多线程实现相同的目的，但线程成本高昂 - 它们需要分配内存，切换它们需要时间等。另一方面，asyncio API 允许您从实际实现中抽象出来，只考虑要执行的作业异步。它的实现可能不同，它包括调用 OS API 和 OS 决定做什么，例如DMA、附加线程、一些特定的微 Controller 使用等。问题是由于较低级别的机制、硬件内容，它对 IO 运行良好。另一方面，执行计算需要将计算算法显式分解为多个部分以用作 asyncio 协程，因此单独的线程可能是更好的决定，因为您可以在那里启动整个计算。 (我不是在谈论并行计算所特有的算法)。但是 asyncio 事件循环可能被明确设置为对协程使用单独的线程，因此这将是具有多线程的 asyncio。

关于您的示例，如果您将使用 sleep 实现您的功能作为 asyncio 协程，调度并运行其中 50 个单线程，您将获得与第一次测试相似的时间，即大约 25s ，因为它正在阻塞。如果您将其更改为类似 yield from [asyncio.sleep][3](0.5) 的内容(这是一个协程本身)，将其中的 50 个单线程运行，它将被异步调用。因此，当一个协程将休眠时，另一个协程将启动，依此类推。作业将在类似于您的第二个多线程测试的时间内完成，即接近 0.5s .如果您要添加 print在这里，您将获得良好的输出，因为它将以串行方式由单线程使用，但输出的顺序可能与协程分配给循环的顺序不同，因为协程可以以不同的顺序运行。如果您将使用多个线程，那么结果显然会接近最后一个。

简化:multithreading 和 asyncio 的区别在于阻塞/非阻塞，所以基本上阻塞多线程会有点接近非阻塞 asyncio，但有很多不同。

用于计算的多线程(即 CPU 绑定(bind)代码)

用于输入/输出的异步(即 I/O 绑定(bind)代码)

关于你的原始陈述:

• all asynchronous code is multi-threaded
• all multi-threaded functions are asynchronous

我希望我能够证明:

• asynchronous code might be both single threaded and multi-threaded
• all multi-threaded functions could be called "asynchronous"