为什么asyncio库比这个I/O绑定操作的线程慢？

Question

我正在编写一个用于枚举网站域名的python程序.例如,'a.google.com'.

首先,我使用该threading模块执行此操作:

import string
import time
import socket
import threading
from threading import Thread
from queue import Queue

'''
enumerate a site's domain name like this:
1-9 a-z + .google.com
1.google.com
2.google.com
.
.
1a.google.com
.
.
zz.google.com

'''

start = time.time()
def create_host(char):
    '''
    if char is '1-9a-z'
    create char like'1,2,3,...,zz'
    '''
    for i in char:
        yield i
    for i in create_host(char):
        if len(i)>1:
            return False
        for c in char:
            yield c + i


char = string.digits + string.ascii_lowercase
site = '.google.com'


def getaddr():
    while True:
        url = q.get()
        try:
            res = socket.getaddrinfo(url,80)
            print(url + ":" + res[0][4][0])
        except:
            pass
        q.task_done()

NUM=1000  #thread's num
q=Queue()

for i in range(NUM):
    t = Thread(target=getaddr)
    t.setDaemon(True)
    t.start()

for host in create_host(char):
    q.put(host+site)
q.join()

end = time.time()

print(end-start)

'''
used time:
9.448670148849487
'''

后来,我读了一本书,说在某些情况下,协同程序比线程更快.所以,我重写了要使用的代码asyncio:

import asyncio
import string
import time


start = time.time()
def create_host(char):
    for i in char:
        yield i
    for i in create_host(char):
        if len(i)>1:
            return False
        for c in char:
            yield c + i


char = string.digits + string.ascii_lowercase
site = '.google.com'

@asyncio.coroutine
def getaddr(loop, url):
    try:
        res = yield from loop.getaddrinfo(url,80)
        print(url + ':' + res[0][4][0])
    except:
        pass

loop = asyncio.get_event_loop()
coroutines = asyncio.wait([getaddr(loop, i+site) for i in create_host(char)])
loop.run_until_complete(coroutines)

end = time.time()

print(end-start)


'''
time 
120.42313003540039
'''

为什么asyncio版本 getaddrinfo这么慢？我是否以某种方式滥用协同程序？

Answer 1

首先,我无法重现几乎与您在Linux机器上看到的性能差异.我一直看到螺纹版本大约需要20-25秒,版本需要24-34秒asyncio.

现在,为什么asyncio慢？有一些事情可以促成这一点.首先,asyncio版本必须按顺序打印,但线程版本不能.打印是I/O,因此GIL可以在发生时释放.这意味着可能有两个或更多线程可以在同一时间打印,但实际上它可能不会经常发生,并且可能不会在性能上产生太大的差异.

其次,更重要的是,该asyncio版本getaddrinfo实际上只调用socket.getaddrinfo在ThreadPoolExecutor:

def getaddrinfo(self, host, port, *,
                family=0, type=0, proto=0, flags=0):
    if self._debug:
        return self.run_in_executor(None, self._getaddrinfo_debug,
                                    host, port, family, type, proto, flags)
    else:
        return self.run_in_executor(None, socket.getaddrinfo,
                                    host, port, family, type, proto, flags)

它使用默认值ThreadPoolExecutor,只有五个线程:

# Argument for default thread pool executor creation.
_MAX_WORKERS = 5

对于这个用例,这并不是你想要的并行性.为了使其行为更像threading版本,您需要使用ThreadPoolExecutor1000个线程,通过以下方式将其设置为默认执行程序loop.set_default_executor:

loop = asyncio.get_event_loop()
loop.set_default_executor(ThreadPoolExecutor(1000))
coroutines = asyncio.wait([getaddr(loop, i+site) for i in create_host(char)])
loop.run_until_complete(coroutines)

现在,这将使行为更加相同threading,但现实情况是你真的没有使用异步I/O - 你只是使用threading不同的API.所以你在这里做的最好的是与threading例子相同的表现.

最后,你并没有真正在每个例子中运行相同的代码 - threading版本使用的是共享a的工作池queue.Queue,而asyncio版本正在为url列表中的每个项目生成一个协程.如果我asyncio使用一个asyncio.Queue和协同池的版本,除了删除print语句和制作一个更大的默认执行程序,我得到两个版本基本相同的性能.这是新asyncio代码:

import asyncio
import string
import time
from concurrent.futures import ThreadPoolExecutor

start = time.time()
def create_host(char):
    for i in char:
        yield i
    for i in create_host(char):
        if len(i)>1:
            return False
        for c in char:
            yield c + i


char = string.digits + string.ascii_lowercase
site = '.google.com'

@asyncio.coroutine
def getaddr(loop, q):
    while True:
        url = yield from q.get()
        if not url:
            break
        try:
            res = yield from loop.getaddrinfo(url,80)
        except:
            pass

@asyncio.coroutine
def load_q(loop, q):
    for host in create_host(char):
        yield from q.put(host+site)
    for _ in range(NUM):
        yield from q.put(None)

NUM = 1000
q = asyncio.Queue()

loop = asyncio.get_event_loop()
loop.set_default_executor(ThreadPoolExecutor(NUM))
coros = [asyncio.async(getaddr(loop, q)) for i in range(NUM)]
loop.run_until_complete(load_q(loop, q))
loop.run_until_complete(asyncio.wait(coros))

end = time.time()

print(end-start)

和每个的输出:

dan@dandesk:~$ python3 threaded_example.py
20.409344911575317
dan@dandesk:~$ python3 asyncio_example.py
20.39924192428589

但请注意,由于网络存在一些可变性.它们有时比这慢几秒.