关于async并发的疑惑

之前一直很好奇像sanic或者aiohttp是如何实现并发的，其实从网上看无非就是使用了asyncio,更底层就是事件驱动，libevent之类的，但是纯从应用层来讲，这个会是怎么一回事呢，举个例子:

async def worker():
    # 模拟消耗时间
    await asyncio.sleep(3)
    return "ok"

async def concurrent():
    # 模拟并发
    tasks = [worker() for _ in range(10)]
    await asyncio.gather(*tasks)
    

从上面代码来看，整个执行周期为3s,但是有个问题是：如果请求一个一个来，如何不阻塞呢？？

async def worker(message):
    # deal message
    await asyncio.sleep(3)
    return "result"

async def consumer():
    # 此处一下过来7条消息
    # 这里如何实现并行执行这7条消息呢？？
    # await worker(message) # ?? 这样会阻塞，并不是一旦发生阻塞就调度出去.
    # 那如果这样呢？
    asyncio.create_task(worker(message))
    # 来了消息创建一个任务，调度出去，不阻塞，这样就可以接收大量消息喽？？

如果按照例如eventlet的做法:

procpool = GreenPool(size=poolsize)

def worker(message):
    pass

def consumer():
    gt = procpool.spawn(worker, message)
    gt.link(handle_message_processed, message) # 此处省略

消息过来，直接抛给pool，并不会进行阻塞,并且有poolsize控制池大小，

额，这里是sanic源码中的一部分…

# -*- coding: utf-8 -*-

"""
-------------------------------------------------
   File Name：     serve
   Description :
   Author :       yu.zhang
   date：          18-7-13
-------------------------------------------------
"""
import asyncio
from functools import partial


class EchoServer(asyncio.Protocol):
    def __init__(self, loop, connections=set()):
        self.connections = connections
        self.transport = None
        self.loop = loop

    def connection_made(self, transport):
        self.connections.add(self)
        self.transport = transport

    def connection_lost(self, exc):
        self.connections.discard(self)
        print('closed:', len(self.connections))

    def data_received(self, data):
        print(data)
        self.transport.write(data)


def serve(host, port, connections=None, protocol=EchoServer, loop=None, reuse_port=False):
    connections = connections if connections is not None else set()
    server = partial(protocol,
                     loop=loop,
                     connections=connections)

    server_coroutine = loop.create_server(server, host, port, reuse_port=reuse_port)
    try:
        echo_server = loop.run_until_complete(server_coroutine)
    except BaseException:
        return
    try:
        loop.run_forever()
    finally:
        for connection in connections:
            print(dir(connection))
        loop.run_until_complete(echo_server.wait_closed())
        loop.close()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    serve('127.0.0.1', 8080, loop=loop)

赏

谢谢你请我吃糖果

支付宝