aleimu / asyncio_different Goto Github PK

"""
_i：子生成器, 同时也是一个迭代器
_y：子生成器生产的值
_r：yield from 表达式最终的值
_s：调用方通过send()发送的值
_e：异常对象
"""
#简化版
_i = iter(EXPR)
try:
    _y = next(_i)
except StopIteration as _e:
    _r = _e.value
else:
    while 1:
        try:
            _s = yield _y
        except StopIteration as _e:
            _r = _e.value
            break
RESULT = _r


#完整版
 _i = iter(EXPR)
try:
    _y = next(_i)
except StopIteration as _e:
    _r = _e.value

else:
    while 1:
        try:
            _s = yield _y
        except GeneratorExit as _e:
            try:
                _m = _i.close
            except AttributeError:
                pass
            else:
                _m()
            raise _e
        except BaseException as _e:
            _x = sys.exc_info()
            try:
                _m = _i.throw
            except AttributeError:
                raise _e
            else:
                try:
                    _y = _m(*_x)
                except StopIteration as _e:
                    _r = _e.value
                    break
        else:
            try:
                if _s is None:
                    _y = next(_i)
                else:
                    _y = _i.send(_s)
            except StopIteration as _e:
                _r = _e.value
                break
RESULT = _r

n = m = 5
flag = "stop"  # 子生成器停止信号,此例子中是有调用者控制,也可以改写成子生成器控制,调用者检查到信号还停止迭代子生成器.

"""
1、调用方：调用委派生成器的客户端（调用方）代码
2、委托生成器：包含yield from 表达式的生成器函数
3、子生成器：yield from 后面加的生成器函数

重点:yield让函数中断执行,next或send让函数恢复执行,使用debug查看各个函数间的跳转,或者直接运行,看print打印.
"""


def gen():  # 子生成器
    print("start 子生成器")
    # for k in range(n):    # 有限子生成器
    k = "k"
    while True:  # 无限子生成器
        print("子生成器--要返回的值:", k)
        t = yield k  # 1.运行到这里就会停下来,切换到其他地方,等待send或next触发后再从此处继续执行 2.yield功能相当于golang中的chan,可接受可发送
        print("子生成器--接受到的值:", t)
        if t is flag:
            break
    print("end 子生成器")
    return "这就是result"  # 生成器退出时, 生成器(或子生成器)中的return expr表达式会出发StopIteration(expr)异常抛出


def proxy_gen():  # 委托生成器--类似go-chan
    # 在调用方与子生成器之间建立一个双向通道,调用方可以通过send()直接发送消息给子生成器,而子生成器yield的值,也是直接返回给调用方
    # while True:
    result = yield from gen()
    print("委托生成器result:", result)
    yield result


def main1():  # 调用方1--不通过proxy_gen迭代子生成器
    g = gen()  # 子生成器
    print(g.send(None))
    print(g.send(1))  # 发送1到子生成器中
    print(next(g))
    try:
        print(g.send(flag))  # 不使用委托器 子生成器的停止信号就得手动处理
    except StopIteration as e:
        print("StopIteration")
        print("子生成器return的值:", e.value)


def main2():  # 调用方2--常用迭代
    g = proxy_gen()
    g.send(None)  # 需要先激活子生成器,否则会报错 TypeError: can't send non-None value to a just-started generator
    for k in range(m):
        print("调用方--要发送的值:", k)
        print("调用方--接受到的值:", g.send(k))
        print("--------------------")
    g.send(flag)  # 针对无限子生成器的停止信号


def main3():  # 调用方3--死循环
    g = proxy_gen()
    g.send(None)  # 需要先激活子生成器,否则会报错 TypeError: can't send non-None value to a just-started generator
    for k in g:  # for调用能完整的遍历生成器,遍历的时候已经调用了__next__,相当于g.send(None)
        print("调用方--接受到的值:", k)
        print("调用方--要发送的值:", g.send("m"))
        print("调用方--接受到的值:", k)
        print("--------------------")


print("*********************")
main1()
print("*********************")
main2()
print("*********************")
main3()
print("*********************")

__doc__ = '如何使用yield完成协程(简化版的asyncio)'

import socket
from selectors import DefaultSelector, EVENT_WRITE, EVENT_READ

selector = DefaultSelector()
stopped = False
host = "127.0.0.1"  # 自建一个简单服务,模拟一个设置每个请求需要等待1s才返回结果
port = 5000
urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}
# urls_todo = {'/'}

# 在单线程内做协作式多任务调度
# 要异步,必回调
# 但为了避免地狱式回调,将回调一拆为三,回调链变成了Future-Task-Coroutine
# 下面的注解都是为了能方便理解Future-Task-Coroutine之间的互动以及怎么串起来的.

"""
无链式调用
selector的回调里只管给future设置值, 不再关心业务逻辑
loop 内回调callback()不再关注是谁触发了事件,因为协程能够保存自己的状态, 知道自己的future是哪个。也不用关心到底要设置什么值, 因为要设置什么值也是协程内安排的。
已趋近于同步代码的结构
无需程序员在多个协程之间维护状态, 例如哪个才是自己的sock
"""

"""
1.创建Crawler 实例；
2.调用fetch方法, 会创建socket连接和在selector上注册可写事件；
3.fetch内并无阻塞操作, 该方法立即返回；
4.重复上述3个步骤, 将10个不同的下载任务都加入事件循环；
5.启动事件循环, 进入第1轮循环, 阻塞在事件监听上；
6.当某个下载任务EVENT_WRITE被触发, 回调其connected方法, 第一轮事件循环结束；
7.进入第2轮事件循环, 当某个下载任务有事件触发, 执行其回调函数；此时已经不能推测是哪个事件发生, 因为有可能是上次connected里的EVENT_READ先被触发, 也可能是其他某个任务的EVENT_WRITE被触发；（此时, 原来在一个下载任务上会阻塞的那段时间被利用起来执行另一个下载任务了）
8.循环往复, 直至所有下载任务被处理完成
9.退出事件循环, 结束整个下载程序
"""


# 异步调用执行完的时候, 就把结果放在它里面。这种对象称之为未来对象。
# 暂存task执行的结果和回调
class Future:
    def __init__(self):
        self.result = None
        self._callbacks = []

    def add_done_callback(self, fn):  # 各阶段的回调
        self._callbacks.append(fn)

    def set_result(self, result):
        self.result = result  # 调用结果,b'http请求的结果字符'
        for fn in self._callbacks:  # 重要,回调函数集
            fn(self)  # Task.step


class Task:
    def __init__(self, coro):
        self.coro = coro  # Crawler(url).fetch()
        f = Future()
        # f.set_result(None)  # 感觉这句不是很必要
        self.step(f)  # 预激活

    def step(self, future):  # 管理fetch生成器: 第一次的激活/暂停后的恢复执行/以及配合set_result循环调用
        try:
            # send会进入到coro执行, 即fetch, 直到下次yield
            # next_future 为yield返回的对象,也就是下一次要调用的Future对象
            next_future = self.coro.send(future.result)  # __init__中的第一次step,将fetch运行到的82行的yield,
            # 返回EVENT_WRITE时的事件回调要用的future,然后等事件触发,由select调用on_connected,进而继续future中的回调
        except StopIteration:
            return
        next_future.add_done_callback(self.step)  # 这里需要重点理解,为下一次要调用的Future对象,注册下一次的step,供on_readable调用


# Coroutine yield实现的协程
class Crawler:
    def __init__(self, url):
        self.url = url
        self.response = b''

    def fetch(self):  # 函数内有了yield表达式,就是生成器了,生成器需要先调用next()迭代一次或者是先send(None)启动,遇到yield之后便暂停
        sock = socket.socket()
        sock.setblocking(False)
        try:
            sock.connect((host, port))
        except BlockingIOError:
            pass
        f = Future()  # 每到一个io事件都注册一个对应的Future

        def on_connected():
            # pass    # 若没有f.set_result,会报错KeyError: '236 (FD 236) is already registered'
            f.set_result(None)  # 必要语句,还涉及到恢复回调

        selector.register(sock.fileno(), EVENT_WRITE, on_connected)  # 连接io写事件
        yield f  # 注册完就yield出去,等待事件触发
        selector.unregister(sock.fileno())
        get = 'GET {0} HTTP/1.0\r\nHost: example.com\r\n\r\n'.format(self.url)  # self.url 区分每个协程
        sock.send(get.encode('ascii'))

        global stopped
        while True:
            f = Future()

            def on_readable():
                f.set_result(sock.recv(4096))  # 可读的情况下,读取4096个bytes暂存给Future,执行回调,使生成器继续执行下去

            selector.register(sock.fileno(), EVENT_READ, on_readable)  # io读事件
            chunk = yield f  # 返回f,并接受step中send进来的future.result值,也就是暂存的请求返回字符
            selector.unregister(sock.fileno())
            if chunk:
                self.response += chunk
            else:
                urls_todo.remove(self.url)
                if not urls_todo:
                    stopped = True
                break
        print("result:", self.response)


def loop():
    while not stopped:
        # 阻塞, 直到一个事件发生
        events = selector.select()
        for event_key, event_mask in events:  # 监听事件,触发回调,推动协程运行下去
            callback = event_key.data  # 就是 on_connected,和 on_readable
            callback()


if __name__ == '__main__':
    import time

    start = time.time()
    for url in urls_todo:
        crawler = Crawler(url)
        Task(crawler.fetch())
    loop()
    print(time.time() - start)

__doc__ = '如何使用yield from完成协程(简化版的asyncio)'

import socket
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE

selector = DefaultSelector()
stopped = False
host = "127.0.0.1"  # 自建一个简单服务,模拟一个设置每个请求需要等待1s才返回结果
port = 5000
urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}
# urls_todo = {'/'}

# 在单线程内做协作式多任务调度
# 要异步,必回调
# 但为了避免地狱式回调,将回调一拆为三,回调链变成了Future-Task-Coroutine
# 下面的注解都是为了能方便理解Future-Task-Coroutine之间的互动以及怎么串起来的.

"""
无链式调用
selector的回调里只管给future设置值, 不再关心业务逻辑
loop 内回调callback()不再关注是谁触发了事件,因为协程能够保存自己的状态, 知道自己的future是哪个。也不用关心到底要设置什么值, 因为要设置什么值也是协程内安排的。
已趋近于同步代码的结构
无需程序员在多个协程之间维护状态, 例如哪个才是自己的sock
"""

"""
1.创建Crawler 实例；
2.调用fetch方法, 会创建socket连接和在selector上注册可写事件；
3.fetch内并无阻塞操作, 该方法立即返回；
4.重复上述3个步骤, 将10个不同的下载任务都加入事件循环；
5.启动事件循环, 进入第1轮循环, 阻塞在事件监听上；
6.当某个下载任务EVENT_WRITE被触发, 回调其connected方法, 第一轮事件循环结束；
7.进入第2轮事件循环, 当某个下载任务有事件触发, 执行其回调函数；此时已经不能推测是哪个事件发生, 因为有可能是上次connected里的EVENT_READ先被触发, 也可能是其他某个任务的EVENT_WRITE被触发；（此时, 原来在一个下载任务上会阻塞的那段时间被利用起来执行另一个下载任务了）
8.循环往复, 直至所有下载任务被处理完成
9.退出事件循环, 结束整个下载程序
"""


# 结果保存, 每一个处需要异步的地方都会调用, 保持状态和callback
# 程序得知道当前所处的状态, 而且要将这个状态在不同的回调之间延续下去。
class Future:
    def __init__(self):
        self.result = None  # 重要参数1
        self._callbacks = []  # 重要参数2

    def add_done_callback(self, fn):  # 各阶段的回调
        self._callbacks.append(fn)

    def set_result(self, result):
        self.result = result  # 调用结果,b'http请求的结果字符'
        for fn in self._callbacks:
            fn(self)  # 执行Task.step

    def __iter__(self):
        """
        yield的出现使得__iter__函数变成一个生成器, 生成器本身就有next方法, 所以不需要额外实现。
        yield from x语句首先调用iter(x)获取一个迭代器（生成器也是迭代器）
        """
        yield self  # 外面使用yield from把f实例本身返回
        return self.result  # 在Task.step中send(result)的时候再次调用这个生成器, 但是此时会抛出stopInteration异常, 并且把self.result返回


# 激活包装的生成器, 以及在生成器yield后恢复执行继续下面的代码
class Task:
    def __init__(self, coro):  # Crawler(url).fetch()
        self.coro = coro
        f = Future()
        # f.set_result(None)
        self.step(f)  # 激活Task包裹的生成器

    def step(self, future):
        try:
            # next_future = self.coro.send(future.result)
            next_future = self.coro.send(None)  # 驱动future
            # next_future = future.send(None)  # 这样是错误的
            # __init__中的第一次step,将fetch运行到的82行的yield,
            # 返回EVENT_WRITE时的事件回调要用的future,然后等事件触发,由select调用on_connected,进而继续future中的回调
        except StopIteration:
            return
        next_future.add_done_callback(self.step)  # 这里需要重点理解,为下一次要调用的Future对象,注册下一次的step,供on_readable调用


# 异步就是可以暂定的函数, 函数间切换的调度靠事件循环,yield 正好可以中断函数运行
# Coroutine yield实现的协程
# 将yield_demo.py中的Crawler进行了拆解,并使用yield from
class Crawler:
    def __init__(self, url):
        self.url = url
        self.response = b""

    def fetch(self):  # 委托生成器,参考yield_to_from.py
        global stopped
        sock = socket.socket()
        yield from connect(sock, (host, port))
        get = "GET {0} HTTP/1.0\r\nHost:example.com\r\n\r\n".format(self.url)
        sock.send(get.encode('ascii'))
        self.response = yield from read_all(sock)
        print(self.response)
        urls_todo.remove(self.url)
        if not urls_todo:
            stopped = True


# 连接事件的子协程:注册+回调
def connect(sock, address):
    f = Future()
    sock.setblocking(False)
    try:
        sock.connect(address)
    except BlockingIOError:
        pass

    def on_connected():
        f.set_result(None)

    selector.register(sock.fileno(), EVENT_WRITE, on_connected)
    yield from f  # f需要可迭代,需要新增Future.__iter__
    selector.unregister(sock.fileno())


# 可读事件的子协程:注册+回调
def read(sock):
    f = Future()

    def on_readable():
        f.set_result(sock.recv(4096))

    selector.register(sock.fileno(), EVENT_READ, on_readable)  # 注册一个文件对象以监听其IO事件;
    """
    此处的chunck接收的是f中return的f.result, 同时会跑出一个stopIteration的异常, 只不过被yield from处理了。
    这里也可直接写成chunck = yiled f
    """
    chunck = yield from f  # f需要可迭代,需要新增Future.__iter__
    selector.unregister(sock.fileno())  # 从selection中注销文件对象, 即从监听列表中移除它; 文件对象应该在关闭前注销.
    return chunck


# 委托生成器,参考yield_to_from.py,生成器的嵌套
def read_all(sock):
    response = []
    chunk = yield from read(sock)
    while chunk:
        response.append(chunk)
        chunk = yield from read(sock)  # yield from来解决生成器里玩生成器的问题
    result = b"".join(response)
    print("result:", result)  # 打印下结果吧
    return result


# 事件驱动, 让所有之前注册的callback运行起来
def loop():
    while not stopped:
        events = selector.select()
        for event_key, event_mask in events:  # 监听事件,触发回调,推动协程运行下去
            callback = event_key.data  # data就是 on_connected,和 on_readable
            callback()


if __name__ == "__main__":
    import time

    start = time.time()
    for url in urls_todo:
        crawler = Crawler(url)
        Task(crawler.fetch())  # 将各生成器和对应的callback注册到事件循环loop中, 并激活生成器
    loop()
    print(time.time() - start)

__doc__ = "使用asyncio"

import asyncio
import aiohttp

host = 'http://127.0.0.1:5000'
urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}

loop = asyncio.get_event_loop()


async def fetch(url):
    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get(url) as response:
            response = await response.read()
            print("result:", response)
            return response


if __name__ == '__main__':
    import time

    start = time.time()
    tasks = [fetch(host + url) for url in urls_todo]
    loop.run_until_complete(asyncio.gather(*tasks))
    print(time.time() - start)

# 事件驱动, 让所有之前注册的callback运行起来
def loop():
    while not stopped:
        events = selector.select()
        for event_key, event_mask in events:  # 监听事件,触发回调,推动协程运行下去
            callback = event_key.data  # data就是 on_connected,和 on_readable
            callback()

def step(self, future):  # 管理fetch生成器: 第一次的激活/暂停后的恢复执行/以及配合set_result循环调用
    try:
        # send会进入到coro执行, 即fetch, 直到下次yield
        # next_future 为yield返回的对象,也就是下一次要调用的Future对象
        next_future = self.coro.send(future.result)  # __init__中的第一次step,将fetch运行到的82行的yield,
        # 返回EVENT_WRITE时的事件回调要用的future,然后等事件触发,由select调用on_connected,进而继续future中的回调
    except StopIteration:
        return
    next_future.add_done_callback(self.step)  # 这里需要重点理解,为下一次要调用的Future对象,注册下一次的step,供on_readable调用



while True:
    f = Future()

    def on_readable():
        f.set_result(sock.recv(4096))  # 可读的情况下,读取4096个bytes暂存给Future,执行回调,使生成器继续执行下去

    selector.register(sock.fileno(), EVENT_READ, on_readable)  # io读事件
    chunk = yield f  # 返回f,并接受step中send进来的future.result值,也就是暂存的请求返回字符
    selector.unregister(sock.fileno())
    if chunk:
        self.response += chunk
    else:
        urls_todo.remove(self.url)
        if not urls_todo:
            stopped = True
        break
print("result:", self.response)