# 模仿collections.abc中的那些（Python3.7目前只是过渡的兼容版，没有具体实现）
def _alias(origin, params, inst=True):
    return _GenericAlias(origin, params, special=True, inst=inst)

T_co = TypeVar('T_co', covariant=True)  # Any type covariant containers.

Iterable = _alias(collections.abc.Iterable, T_co)
Iterator = _alias(collections.abc.Iterator, T_co)

// 能不能foreach就看你遍历对象有没有实现IEnumerable，就说明你是否是一个可枚举类型
public interface IEnumerable
{
    IEnumerator GetEnumerator();
}

// 是否是个枚举器（enumerator）就看你实现了IEnumerator接口没
public interface IEnumerator
{
    object Current { get; }

    bool MoveNext();

    void Reset();
}

# https://github.com/lotapp/cpython3/blob/master/Lib/_collections_abc.py
class Iterable(metaclass=ABCMeta):

    __slots__ = ()

    @abstractmethod
    def __iter__(self):
        while False:
            yield None

    @classmethod
    def __subclasshook__(cls, C):
        if cls is Iterable:
            return _check_methods(C, "__iter__")
        return NotImplemented

class Iterator(Iterable):

    __slots__ = ()

    @abstractmethod
    def __next__(self):
        'Return the next item from the iterator. When exhausted, raise StopIteration'
        raise StopIteration

    def __iter__(self):
        return self

    @classmethod
    def __subclasshook__(cls, C):
        if cls is Iterator:
            return _check_methods(C, '__iter__', '__next__')
        return NotImplemented

class Generator(Iterator):
    __slots__ = ()

    def __next__(self):
        """从生成器返回下一个item，结束的时候抛出 StopIteration"""
        return self.send(None)

    @abstractmethod
    def send(self, value):
        """将值发送到生成器。返回下一个产生的值或抛出StopIteration"""
        raise StopIteration

    @abstractmethod
    def throw(self, typ, val=None, tb=None):
        """在生成器中引起异常。返回下一个产生的值或抛出StopIteration"""
        if val is None:
            if tb is None:
                raise typ
            val = typ()
        if tb is not None:
            val = val.with_traceback(tb)
        raise val

    # 如今知道以前close后为啥没异常了吧～
    def close(self):
        """屏蔽异常"""
        try:
            self.throw(GeneratorExit)
        except (GeneratorExit, StopIteration):
            pass
        else:
            raise RuntimeError("generator ignored GeneratorExit")

    @classmethod
    def __subclasshook__(cls, C):
        if cls is Generator:
            return _check_methods(C, '__iter__', '__next__',
                                  'send', 'throw', 'close')
        return NotImplemented

def test_send(n):
    for i in range(n):
        if i==2:
            return "i==2"
        yield i

g = test_send(5)

while True:
    try:
        tmp = next(g)
        print(tmp)
    except StopIteration as ex:
        print(ex.value)
        break

import time
import socket

def select(socket_addr_list):
    for client_socket, client_addr in socket_addr_list:
        try:
            data = client_socket.recv(2048)
            if data:
                print(f"[来自{client_addr}的消息：]\n")
                print(data.decode("utf-8"))
                client_socket.send(
                    b"HTTP/1.1 200 ok\r\nContent-Type: text/html;charset=utf-8\r\n\r\n<h1>Web Server Test</h1>"
                )
            else:
                # 没有消息是触发异常，空消息是断开链接
                client_socket.close()  # 关闭客户端链接
                socket_addr_list.remove((client_socket, client_addr))
                print(f"[客户端{client_addr}已断开链接，当前链接数：{len(socket_addr_list)}]")
        except Exception:
            pass

def main():
    # 存放客户端集合
    socket_addr_list = list()

    with socket.socket() as tcp_server:
        # 防止端口绑定的设置
        tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        tcp_server.bind(('', 8080))
        tcp_server.listen()
        tcp_server.setblocking(False)  # 服务端非阻塞
        while True:
            try:
                client_socket, client_addr = tcp_server.accept()
                client_socket.setblocking(False)  # 客户端非阻塞
                socket_addr_list.append((client_socket, client_addr))
            except Exception:
                pass
            else:
                print(f"[来自{client_addr}的链接，当前链接数：{len(socket_addr_list)}]")
            # 防止客户端断开后出错
            if socket_addr_list:
                # 轮询查看客户端有没有消息
                select(socket_addr_list)  # 引用传参
                time.sleep(0.01)

if __name__ == "__main__":
    main()

import select
import socket

def main():
    with socket.socket() as tcp_server:
        tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        tcp_server.bind(('', 8080))
        tcp_server.listen()
        socket_info_dict = dict()
        socket_list = [tcp_server]  # 监测列表
        while True:
            # 劣势：select列表数量有限制
            read_list, write_list, error_list = select.select(
                socket_list, [], [])
            for item in read_list:
                # 服务端迎接新的链接
                if item == tcp_server:
                    client_socket, client_address = item.accept()
                    socket_list.append(client_socket)
                    socket_info_dict[client_socket] = client_address
                    print(f"[{client_address}已链接，当前链接数：{len(socket_list)-1}]")
                # 客户端发来
                else:
                    data = item.recv(2048)
                    if data:
                        print(data.decode("utf-8"))
                        item.send(
                            b"HTTP/1.1 200 ok\r\nContent-Type: text/html;charset=utf-8\r\n\r\n<h1>Web Server Test</h1>"
                        )
                    else:
                        item.close()
                        socket_list.remove(item)
                        info = socket_info_dict[item]
                        print(f"[{info}已断开，当前链接数：{len(socket_list)-1}]")

if __name__ == "__main__":
    main()

import socket
import select

def main():
    with socket.socket() as tcp_server:
        tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        tcp_server.bind(('', 8080))
        tcp_server.listen()

        # epoll是linux独有的
        epoll = select.epoll()
        # tcp_server注册到epoll中
        epoll.register(tcp_server.fileno(), select.EPOLLIN | select.EPOLLET)

        # key-value
        fd_socket_dict = dict()

        # 回调须要本身处理
        while True:
            # 返回可读写的socket fd 集合
            poll_list = epoll.poll()
            for fd, event in poll_list:
                # 服务器的socket
                if fd == tcp_server.fileno():
                    client_socket, client_addr = tcp_server.accept()
                    fd = client_socket.fileno()
                    fd_socket_dict[fd] = (client_socket, client_addr)
                    # 把客户端注册进epoll中
                    epoll.register(fd, select.EPOLLIN | select.EPOLLET)
                else:  # 客户端
                    client_socket, client_addr = fd_socket_dict[fd]
                    data = client_socket.recv(2048)
                    print(
                        f"[来自{client_addr}的消息，当前链接数：{len(fd_socket_dict)}]\n")
                    if data:
                        print(data.decode("utf-8"))
                        client_socket.send(
                            b"HTTP/1.1 200 ok\r\nContent-Type: text/html;charset=utf-8\r\n\r\n<h1>Web Server Test</h1>"
                        )
                    else:
                        del fd_socket_dict[fd]
                        print(
                            f"[{client_addr}已离线，当前链接数：{len(fd_socket_dict)}]\n"
                        )
                        # 从epoll中注销
                        epoll.unregister(fd)
                        client_socket.close()

if __name__ == "__main__":
    main()

# 选择级别：epoll|kqueue|devpoll > poll > select
if 'KqueueSelector' in globals():
    DefaultSelector = KqueueSelector
elif 'EpollSelector' in globals():
    DefaultSelector = EpollSelector
elif 'DevpollSelector' in globals():
    DefaultSelector = DevpollSelector
elif 'PollSelector' in globals():
    DefaultSelector = PollSelector
else:
    DefaultSelector = SelectSelector

import socket
import selectors

# Linux下使用epoll，Win下使用select
Selector = selectors.DefaultSelector()

class Task(object):
    def __init__(self):
        # 存放客户端fd和socket键值对
        self.fd_socket_dict = dict()

    def run(self):
        self.server = socket.socket()
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server.bind(('', 8080))
        self.server.listen()
        # 把Server注册到epoll
        Selector.register(self.server.fileno(), selectors.EVENT_READ,
                          self.connected)

    def connected(self, key):
        """客户端链接时处理"""
        client_socket, client_address = self.server.accept()
        fd = client_socket.fileno()
        self.fd_socket_dict[fd] = (client_socket, client_address)
        # 注册一个客户端读的事件（服务端去读消息）
        Selector.register(fd, selectors.EVENT_READ, self.call_back_reads)
        print(f"{client_address}已链接，当前链接数：{len(self.fd_socket_dict)}")

    def call_back_reads(self, key):
        """客户端可读时处理"""
        # 一个fd只能注册一次，监测可写的时候须要把可读给注销
        Selector.unregister(key.fd)
        client_socket, client_address = self.fd_socket_dict[key.fd]
        print(f"[来自{client_address}的消息:]\n")
        data = client_socket.recv(2048)
        if data:
            print(data.decode("utf-8"))
            # 注册一个客户端写的事件（服务端去发消息）
            Selector.register(key.fd, selectors.EVENT_WRITE,
                              self.call_back_writes)
        else:
            client_socket.close()
            del self.fd_socket_dict[key.fd]
            print(f"{client_address}已断开，当前链接数：{len(self.fd_socket_dict)}")

    def call_back_writes(self, key):
        """客户端可写时处理"""
        Selector.unregister(key.fd)
        client_socket, client_address = self.fd_socket_dict[key.fd]
        client_socket.send(b"ok")
        Selector.register(key.fd, selectors.EVENT_READ, self.call_back_reads)

def main():
    t = Task()
    t.run()
    while True:
        ready = Selector.select()
        for key, obj in ready:
            # 须要本身回调
            call_back = key.data
            call_back(key)

if __name__ == "__main__":
    main()

# https://docs.python.org/3/library/itertools.html#itertools.chain
import itertools

def main():
    # 模拟分表后的两个查询结果
    user1 = ["小张", "小明"]
    user2 = ["小潘", "小周"]
    # dict只能遍历key（这种状况须要本身封装合并方法并处理下）
    user3 = {"name": "test1", "name1": "test2"}

    # 需求：合并并遍历
    for item in itertools.chain(user1, user2, user3):
        print(item)

if __name__ == '__main__':
    main()

def my_chain(*args, **kwargs):
    for items in args:
        for item in items:
            yield item

def main():
    # 模拟分表后的两个查询结果
    user1 = ["小张", "小明"]
    user2 = ["小潘", "小周"]
    # dict只能遍历key（这种状况须要本身封装合并方法并处理下）
    user3 = {"name": "test1", "name1": "test2"}

    # 需求：合并并遍历
    for item in my_chain(user1, user2, user3):
        print(item)

if __name__ == '__main__':
    main()

def my_chain(*args, **kwargs):
    for items in args:
        yield from items

def main():
    # 模拟分表后的两个查询结果
    user1 = ["小张", "小明"]
    user2 = ["小潘", "小周"]

    # 需求：合并并遍历
    for item in my_chain(user1, user2):
        print(item)

if __name__ == '__main__':
    main()

def my_chain(*args, **kwargs):
    for my_iterable in args:
        # 若是是字典类型就返回value
        if isinstance(my_iterable, dict):
            my_iterable = my_iterable.values()
        for item in my_iterable:
            yield item

def main():
    # 模拟分表后的两个查询结果
    user1 = ["小张", "小明"]
    user2 = ["小潘", "小周"]
    # dict只能遍历key（这种状况须要本身封装合并方法并处理下）
    user3 = {"name": "test1", "name1": "test2"}
    # 需求：合并并遍历
    for item in my_chain(user1, user2, user3):
        print(item)

if __name__ == '__main__':
    main()

import itertools

def main():
    # 模拟分表后的两个查询结果
    user1 = [{"name": "小张"}, {"name": "小明"}]
    user2 = [{"name": "小潘"}, {"name": "小周"}]
    user3 = [{"name": "test1"}, {"name": "test2"}]
    # 需求：合并并遍历
    for item in itertools.chain(user1, user2, user3):
        # 通常都是直接在这里进行处理
        for key, value in item.items():
            print(value)

if __name__ == '__main__':
    main()

products = [{
    "id": 2344,
    "title": "御泥坊补水面膜",
    "price": [89, 76, 120, 99]
}, {
    "id": 2345,
    "title": "御泥坊火山泥面膜",
    "price": [30, 56, 70, 89]
}]

new_products = [{
    "id": 2344,
    "title": "御泥坊补水面膜",
    "price": [89, 76, 120, 99],
    "max": 120,
    "min": 76,
    "avg": 96.0
},
{
    "id": 2345,
    "title": "御泥坊火山泥面膜",
    "price": [30, 56, 70, 89],
    "max": 89,
    "min": 30,
    "avg": 61.25
}]

# 正常调用
RESULT = yield from EXPR

# _i：子生成器（也是个迭代器）
# _y：子生成器生产的值
# _r：yield from 表达式最终结果
# _s：调用方经过send发送的值
# _e：异常对象

# 内部原理
_i = iter(EXPR) # EXPR是一个可迭代对象，_i是子生成器
try:
    # 第一次不能send值，只能next() or send(None)，并把产生的值放到_y中
    _y = next(_i)
except StopIteration as _e:
    # 若是子生成器直接就return了，那就会抛出异常，经过value能够拿到子生成器的返回值
    _r = _e.value
else:
    # 尝试进行循环（调用方和子生成器交互过程），yield from这个生成器会阻塞（委托生成器）
    while 1:
        # 这时候子生成器已经和调用方创建了双向通道，在等待调用方send(value)，把这个值保存在_s中
        _s = yield _y # 这边还会进行一系列异常处理，我先删掉，等会看
        try:
            # 若是send(None)，那么继续next遍历
            if _s is None:
                _y = next(_i) # 把子生成器结果放到 _y 中
            else:
                _y = _i.send(_s) # 若是调用方send一个值，就转发到子生成器
        except StopIteration as _e:
            _r = _e.value # 若是子生成器遍历完了，就把返回值给_r
            break
RESULT = _r # 最终的返回值（yield from 最终的返回值）

# 正常调用
RESULT = yield from EXPR

# _i：子生成器（也是个迭代器）
# _y：子生成器生产的值
# _r：yield from 表达式最终结果
# _s：调用方经过send发送的值
# _e：异常对象

# 内部原理
_i = iter(EXPR) # EXPR是一个可迭代对象，_i是子生成器
try:
    # 第一次不能send值，只能next() or send(None)，并把产生的值放到_y中
    _y = next(_i)
except StopIteration as _e:
    # 若是子生成器直接就return了，那就会抛出异常，经过value能够拿到子生成器的返回值
    _r = _e.value
else:
    # 尝试进行循环（调用方和子生成器交互过程），yield from这个生成器会阻塞（委托生成器）
    while 1:
        try:
            # 这时候子生成器已经和调用方创建了双向通道，在等待调用方send(value)，把这个值保存在_s中
            _s = yield _y

        # 【如今补全】有这么几种状况须要处理
        # 1.子生成器可能只是一个迭代器，并不能做为协程的生成器（不支持throw和close）
        # 2.子生成器虽然支持了throw和close，但在子生成器内部两种方法都会抛出异常
        # 3.调用法调用了gen.throw()，想让子生成器本身抛异常
        # 这时候就要处理 gen.close() 和 gen.throw()的状况

        # 生成器close()异常的处理
        except GeneratorExit as _e:
            try:
                _m = _i.close
            except AttributeError:
                pass # 屏蔽close的异常
            else:
                _m()
            raise _e # 上抛异常
        # 生成器throw()异常的处理
        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:
                # 若是send(None)，那么继续next遍历
                if _s is None:
                    _y = next(_i) # 把子生成器结果放到 _y 中
                else:
                    _y = _i.send(_s) # 若是调用方send一个值，就转发到子生成器
            except StopIteration as _e:
                _r = _e.value # 若是子生成器遍历完了，就把返回值给_r
                break
RESULT = _r # 最终的返回值（yield from 最终的返回值）

import asyncio

# 生成新的dict数据
async def get_new_item(item):
    prices = item["price"]
    item["avg"] = sum(prices) / len(prices)
    item["max"] = max(prices)
    item["min"] = min(prices)
    return item

async def get_new_data(data):
    newdata = []
    for item in data:
        new_item = await get_new_item(item)
        # print(new_item) # 处理后的新dict
        newdata.append(new_item)
    return newdata

def main():
    # 需求：生成绘图的数据（max,min,avg）
    products = [{
        "id": 2344,
        "title": "御泥坊补水面膜",
        "price": [89, 76, 120, 99]
    }, {
        "id": 2345,
        "title": "御泥坊火山泥面膜",
        "price": [30, 56, 70, 89]
    }]

    # python 3.7
    new_products = asyncio.run(get_new_data(products))
    print(new_products)

if __name__ == "__main__":
    main()

import types

# 和生成器彻底分开了，不过能够理解为yield from
@types.coroutine
def get_value(value):
    yield value

async def get_name(name):
    # 一系列逻辑处理
    return await get_value(name)

if __name__ == '__main__':
    gen = get_name("小明")
    print(gen.send(None))
# 直接混用会报错：TypeError: object generator can't be used in 'await' expression

import types

def get_value(value):
    yield value

# 这个async和await替换成yield from
def get_name(name):
    # 一系列逻辑处理
    yield from get_value(name)

if __name__ == '__main__':
    gen = get_name("小明")
    print(gen.send(None))

import types
import asyncio

# 模拟一个耗时操做
async def test():
    print("start...")
    # 不能再使用之前阻塞的暂停了
    await asyncio.sleep(2)
    print("end...")
    return "ok"

if __name__ == '__main__':
    import time
    start_time = time.time()

    # # >=python3.4
    # # 返回asyncio的事件循环
    # loop = asyncio.get_event_loop()
    # # 运行事件循环，直到指定的future运行完毕，返回结果
    # result = loop.run_until_complete(test())
    # print(result)

    # python3.7
    result = asyncio.run(test())
    print(result)

    print(time.time() - start_time)

# https://github.com/lotapp/cpython3/blob/master/Lib/asyncio/runners.py
def run(main, *, debug=False):
    # 之前是直接使用"asyncio.get_event_loop()"（开发人员通常都习惯这个了）
    # 3.7开始推荐使用"asyncio.get_running_loop()"来获取正在运行的loop（获取不到就抛异常）
    if events._get_running_loop() is not None:
        raise RuntimeError("没法从正在运行的事件循环中调用asyncio.run()")

    if not coroutines.iscoroutine(main):
        raise ValueError("{!r}应该是一个协程".format(main))

    loop = events.new_event_loop() # 建立一个新的事件循环
    try:
        events.set_event_loop(loop)  # 设置事件循环
        loop.set_debug(debug)  # 是否调试运行（默认否）
        return loop.run_until_complete(main)  # 等待运行
    finally:
        try:
            _cancel_all_tasks(loop)  # 取消其余任务
            loop.run_until_complete(loop.shutdown_asyncgens())
        finally:
            events.set_event_loop(None)
            loop.close()

import asyncio

# 模拟一个耗时操做
async def test(i):
    print("start...")
    # 不能再使用之前阻塞的暂停了
    await asyncio.sleep(2)
    print("end...")
    return i

if __name__ == '__main__':
    import time

    start_time = time.time()

    # # >=python3.4
    loop = asyncio.get_event_loop()
    # tasks = [asyncio.ensure_future(test(i)) for i in range(10)]
    # 注意：是loop的方法，而不是asyncio的，否则就会引起RuntimeError：no running event loop
    tasks = [loop.create_task(test(i)) for i in range(10)]
    loop.run_until_complete(asyncio.wait(tasks))
    for task in tasks:
        print(task.result())

    print(time.time() - start_time)

import asyncio

# 模拟一个耗时操做
async def test(i):
    print("start...")
    # 不能再使用之前阻塞的暂停了
    await asyncio.sleep(2)
    print("end...")
    return i

async def main():
    tasks = [test(i) for i in range(10)]
    # await task 能够获得返回值（获得结果或者异常）
    # for task in asyncio.as_completed(tasks):
    #     try:
    #         print(await task)
    #     except Exception as ex:
    #         print(ex)
    return [await task for task in asyncio.as_completed(tasks)]

if __name__ == '__main__':
    import time

    start_time = time.time()

    # old推荐使用
    loop = asyncio.get_event_loop()
    result_list = loop.run_until_complete(main())
    print(result_list)

    print(time.time() - start_time)

import asyncio

# 模拟一个耗时操做
async def test(i):
    print("start...")
    await asyncio.sleep(2)
    print("end...")
    return i

async def main():
    tasks = [test(i) for i in range(10)]
    # 给`协程/futures`返回一个future聚合结果
    return await asyncio.gather(*tasks) # 记得加*来解包

if __name__ == '__main__':
    import time

    start_time = time.time()

    # python3.7
    result_list = asyncio.run(main())
    print(result_list)

    # 2.0259485244750977
    print(time.time() - start_time)

# 给 协程/futures 返回一个future聚合结果
def gather(*coros_or_futures, loop=None, return_exceptions=False):
    pass

# 把协程或者awaitable对象包裹成task
def ensure_future(coro_or_future, *, loop=None):
    pass

# 传入一个协程对象，返回一个task对象
class BaseEventLoop(events.AbstractEventLoop):
    def create_task(self, coro):
        pass

# 低级API示例
import asyncio

async def get_html(url):
    print(f"get {url} ing")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

def call_back(task):
    print(type(task))
    print(task.result())

if __name__ == "__main__":
    import time
    start_time = time.time()

    urls = [
        "https://www.baidu.com", "https://www.sogou.com",
        "https://www.python.org", "https://www.asp.net"
    ]
    tasks = set()  # 任务集合
    loop = asyncio.get_event_loop()
    for url in urls:
        # task = asyncio.ensure_future(get_html(url))
        task = loop.create_task(get_html(url))
        # 设置回调函数
        task.add_done_callback(call_back)
        # 添加到任务集合中
        tasks.add(task)
    # 批量执行
    loop.run_until_complete(asyncio.gather(*tasks))

    print(time.time() - start_time)

import asyncio
import functools

async def get_html(url):
    await asyncio.sleep(2)
    return "This is a test for"

# 注意一个东西：经过偏函数传过来的参数在最前面
def call_back(url, task):
    # do something
    print(type(task))
    print(task.result(), url)

if __name__ == "__main__":
    import time
    start_time = time.time()

    urls = [
        "https://www.baidu.com", "https://www.sogou.com",
        "https://www.python.org", "https://www.asp.net"
    ]
    tasks = set()  # 任务集合
    loop = asyncio.get_event_loop()
    for url in urls:
        # task = asyncio.ensure_future(get_html(url))
        task = loop.create_task(get_html(url))
        # 设置回调函数 （不支持传参数，咱们就利用偏函数来传递）
        task.add_done_callback(functools.partial(call_back, url))
        # 添加到任务集合中
        tasks.add(task)
    # 批量执行
    loop.run_until_complete(asyncio.gather(*tasks))

    print(time.time() - start_time)

import asyncio

async def get_html(url):
    print(f"get {url} ing")
    if url == "https://www.asp.net":
        raise Exception("Exception is over")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

async def main():
    urls = [
        "https://www.baidu.com", "https://www.asp.net",
        "https://www.python.org", "https://www.sogou.com"
    ]
    tasks = [get_html(url) for url in urls]
    return await asyncio.gather(*tasks)

if __name__ == "__main__":
    import time
    start_time = time.time()

    try:
        asyncio.run(main())
    except Exception as ex:
        print(ex)

    print(time.time() - start_time)

import asyncio

async def get_html(url):
    print(f"get {url} ing")
    if url == "https://www.asp.net":
        raise Exception("Exception is over")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

async def main():
    urls = [
        "https://www.baidu.com", "https://www.asp.net",
        "https://www.python.org", "https://www.sogou.com"
    ]
    tasks = set()  # 任务集合
    tasks = [get_html(url) for url in urls]
    return await asyncio.gather(*tasks)

if __name__ == "__main__":
    import time
    start_time = time.time()

    loop = asyncio.get_event_loop()
    try:
        # 批量执行
        loop.run_until_complete(main())
    except Exception as ex:
        print(ex)

    print(time.time() - start_time)

# 若是和旧版本混用，就应该这么写了（麻烦）
try:
    loop = asyncio.get_running_loop()
except RuntimeError as ex:
    print(ex) # no running event loop
    loop = asyncio.get_event_loop()
...
loop.run_until_complete(xxx)

async def main():
    loop = asyncio.get_running_loop()
    ...

asyncio.run(main())

import asyncio

async def compute(x, y):
    print(f"计算 {x}+{y}...")
    await asyncio.sleep(1.0)
    return x + y

async def main(x, y):
    result = await compute(x, y)
    print(f"{x}+{y}={result}")

loop = asyncio.get_event_loop()
loop.run_until_complete(main(1, 2))
loop.close()

import asyncio

async def get_html(url):
    print(f"get {url} ing")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

def callback_func(task):
    print(type(task))
    if task.done():
        print(f"done")  # print(task.result())

async def main():
    urls = [
        "https://www.baidu.com", "https://www.asp.net",
        "https://www.python.org", "https://www.sogou.com"
    ]
    # asyncio.create_task来建立一个Task
    tasks = [asyncio.create_task(get_html(url)) for url in urls]
    # 给每一个任务都加一个回调函数
    for task in tasks:
        task.add_done_callback(callback_func)
    # 批量执行任务
    result = await asyncio.gather(*tasks)
    print(result)  # 返回 result list

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def get_html(url):
    print(f"get {url} ing")
    if url == "https://www.asp.net":
        raise Exception("Exception is over")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

def callback_func(task):
    if task.done():
        print(f"done")  # print(task.result())

async def main():
    urls = [
        "https://www.baidu.com", "https://www.asp.net",
        "https://www.python.org", "https://www.sogou.com"
    ]
    # asyncio.create_task来建立一个Task
    tasks = [asyncio.create_task(get_html(url)) for url in urls]
    # 给每一个任务都加一个回调函数
    for task in tasks:
        task.add_done_callback(callback_func)
    # 批量执行任务
    result = await asyncio.gather(*tasks, return_exceptions=True)
    print(result)  # 返回 result list

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def get_html(url):
    print(f"get url for{url}")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

async def main():
    urls1 = ["https://www.baidu.com", "https://www.asp.net"]
    urls2 = ["https://www.python.org", "https://www.sogou.com"]

    tasks1 = [asyncio.create_task(get_html(url)) for url in urls1]
    tasks2 = [asyncio.create_task(get_html(url)) for url in urls2]

    # 等待两组都完成，而后返回聚合结果
    result = await asyncio.gather(*tasks1, *tasks2)
    print(result)

if __name__ == "__main__":
    import time
    start_time = time.time()

    try:
        asyncio.run(main())
    except Exception as ex:
        print(ex)

    print(time.time() - start_time)

import asyncio

async def get_html(url):
    print(f"get url for{url}")
    await asyncio.sleep(2)
    return f"<h1>This is a test for {url}</h1>"

async def main():
    urls1 = ["https://www.baidu.com", "https://www.asp.net"]
    urls2 = ["https://www.python.org", "https://www.sogou.com"]

    tasks1 = [asyncio.create_task(get_html(url)) for url in urls1]
    tasks2 = [asyncio.create_task(get_html(url)) for url in urls2]

    group1 = asyncio.gather(*tasks1)
    group2 = asyncio.gather(*tasks2)

    # 分组2由于某缘由被取消任务了（模拟）
    group2.cancel()

    # 等待两组都完成，而后返回聚合结果
    result = await asyncio.gather(group1, group2, return_exceptions=True)
    print(result)

if __name__ == "__main__":
    import time
    start_time = time.time()

    try:
        asyncio.run(main())
    except Exception as ex:
        print(ex)

    print(time.time() - start_time)

import asyncio

async def test():
    print("start...")
    await asyncio.sleep(10)
    print("end...")

async def main():
    task = asyncio.create_task(test())

    await asyncio.sleep(1)

    # 取消task任务
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print(f"任务已经被取消：{task.cancelled()}")
        print(f"任务是由于异常而完成：{task.done()}")

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

# 完成包含了正常+异常
if outer.done():
    # 把由于异常完成的任务打个标记
    if not fut.cancelled():
        fut.exception() # 标记检索的异常

import asyncio

async def test(time):
    print("start...")
    await asyncio.sleep(time)
    print("end...")
    return time

async def main():
    task = asyncio.create_task(test(3))
    try:
        result = await asyncio.wait_for(task, timeout=2)
        print(result)
    except asyncio.CancelledError:
        print("Cancel")
    except asyncio.TimeoutError:
        print("超时取消")
    except Exception as ex:
        print(ex)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def test(time):
    print("start...")
    await asyncio.sleep(time)
    print("end...")
    return time

async def main():
    tasks = [asyncio.create_task(test(i)) for i in range(10)]

    # 已完成的任务（包含异常），未完成的任务
    done, pending = await asyncio.wait(tasks, timeout=2)
    # 任务总数（我用了3种表示）PS：`all_tasks()`的时候记得去除main的那个
    print(
        f"任务总数：{len(tasks)}=={len(done)+len(pending)}=={len(asyncio.Task.all_tasks())-1}"
    )
    # 全部未完成的task：asyncio.all_tasks()，记得去掉run(main())
    print(f"未完成Task:{len(pending)}=={len(asyncio.all_tasks()) - 1}")

    print(await asyncio.gather(*done))
    # for task in done:
        # print(await task)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def test(i):
    print(f"start...task{i}")
    await asyncio.sleep(i)
    print(f"end...task{i}")
    return "ok"

# 第一个任务执行完成则结束此批次任务
async def main():
    tasks = [asyncio.create_task(test(i)) for i in range(10)]

    # 项目里常常有这么一个场景：同时调用多个同效果的API，有一个返回后取消其余请求
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED)

    # print(await asyncio.gather(*done))
    for task in done:
        print(await task)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def test():
    print("start ...")
    await asyncio.sleep(2)
    print("end ...")

# 若是你用`get_running_loop`就不要和`loop.run_until_complete`混用
loop = asyncio.get_event_loop()
loop.run_until_complete(test())

# 若是和旧版本混用，就应该这么写了（麻烦）
try:
    loop = asyncio.get_running_loop()
except RuntimeError as ex:
    loop = asyncio.get_event_loop()
...
asyncio.run(test())

async def main():
    loop = asyncio.get_running_loop()
    ...

asyncio.run(main())

# https://github.com/lotapp/cpython3/blob/master/Lib/asyncio/runners.py
def run(main, *, debug=False):
    if events._get_running_loop() is not None:
        raise RuntimeError("没法从正在运行的事件循环中调用asyncio.run()")

    if not coroutines.iscoroutine(main):
        raise ValueError("{!r}应该是一个协程".format(main))

    # 建立一个新的事件循环
    loop = events.new_event_loop()
    try:
        events.set_event_loop(loop) # 设置事件循环
        loop.set_debug(debug) # 是否调试运行（默认否）
        return loop.run_until_complete(main) # 等待运行
    finally:
        try:
            _cancel_all_tasks(loop) # 取消其余任务
            loop.run_until_complete(loop.shutdown_asyncgens())
        finally:
            events.set_event_loop(None)
            loop.close()

import asyncio
import concurrent.futures

# 模拟一个耗时操做
def test(n):
    return sum(i * i for i in range(10**n))

# old main
def main():
    with concurrent.futures.ThreadPoolExecutor() as pool:
        # 注意：future和asyncio.future是不同的
        future = pool.submit(test, 7)
        result = future.result()
        print(result)

if __name__ == "__main__":
    import time

    start_time = time.time()

    main()  # old

    print(time.time() - start_time)

import asyncio
import concurrent.futures


# 模拟一个耗时操做
def test(n):
    return sum(i * i for i in range(10**n))

async def main():
    # 获取loop
    loop = asyncio.get_running_loop()

    with concurrent.futures.ThreadPoolExecutor() as pool:
        # 新版兼任代码
        result = await loop.run_in_executor(pool, test, 7)
        print(result)


if __name__ == "__main__":
    import time

    start_time = time.time()

    asyncio.run(main())  # new

    print(time.time() - start_time)

class BaseEventLoop(events.AbstractEventLoop):
    def run_in_executor(self, executor, func, *args):
        # 检查loop是否关闭，若是关闭就抛`RuntimeError`异常
        self._check_closed()
        if self._debug:
            self._check_callback(func, 'run_in_executor')
        # 若是不传一个executor,就会使用默认的executor
        # 换句话说：你能够不传`线程池`
        if executor is None:
            executor = self._default_executor
            if executor is None:
                executor = concurrent.futures.ThreadPoolExecutor()
                self._default_executor = executor
        # 把`concurrent.futures.Future`对象封装成`asyncio.futures.Future`对象
        return futures.wrap_future(executor.submit(func, *args), loop=self)

import asyncio

# 模拟一个耗时操做
def test(n):
    return sum(i * i for i in range(10**n))

async def main():
    # 获取loop
    loop = asyncio.get_running_loop()

    # 新版兼任代码
    result = await loop.run_in_executor(None, test, 7)
    print(result)

if __name__ == "__main__":
    import time

    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

def test(name):
    print(f"start {name}...")
    print(f"end {name}...")

async def main():
    # 正在执行某个任务
    loop = asyncio.get_running_loop()

    # 插入一个更要紧的任务
    # loop.call_later(0, callback, *args)
    task1 = loop.call_soon(test, "task1")

    # 多少秒后执行
    task2 = loop.call_later(2, test, "task2")

    # 内部时钟时间
    task3 = loop.call_at(loop.time() + 3, test, "task3")

    print(type(task1))
    print(type(task2))
    print(type(task3))

    # 保证loop在执行完毕后才关闭
    await asyncio.sleep(5)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

# 回调函数通常都是普通函数
def test(name):
    print(name)

if __name__ == "__main__":
    import time
    start_time = time.time()

    loop = asyncio.get_event_loop()

    # 新版本限制了时间不能超过24h（防止有些人当定时任务来乱用）
    # 这个执行过程会安装时间排个前后顺序，而后再批次运行
    task4 = loop.call_later(4, test, "task2-4")
    task2 = loop.call_later(2, test, "task2-2")
    task3 = loop.call_later(3, test, "task2-3")
    task1 = loop.call_later(1, test, "task2-1")
    # 取消测试
    task4.cancel()
    # close是直接丢弃任务而后关闭loop
    loop.call_later(4, loop.stop)  # 等任务执行完成结束任务 loop.stop()

    # run内部运行的是run_until_complete，而run_until_complete内部运行的是run_forever
    loop.run_forever()
    print(time.time() - start_time)

import asyncio

# 不是协程就加个装饰器
@asyncio.coroutine
def test():
    print("this is a test")

async def test_async():
    print("this is a async test")
    await asyncio.sleep(1)

async def main():
    # 传入一个协程对象，返回一个task
    task1 = asyncio.create_task(test())
    task2 = asyncio.create_task(test_async())
    await asyncio.gather(task1, task2)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

# 传入一个协程对象，返回一个Task对象
def create_task(self, coro):
    self._check_closed()
    if self._task_factory is None:
        # look：核心点
        task = tasks.Task(coro, loop=self)
        if task._source_traceback:
            del task._source_traceback[-1]
    else:
        task = self._task_factory(self, coro)
    return task

class Task(futures._PyFuture):
    def __init__(self, coro, *, loop=None):
        super().__init__(loop=loop)
        ...
        # 安排了一个尽快执行的回调方法：self.__step
        self._loop.call_soon(self.__step, context=self._context)

    def __step(self, exc=None):
       try:
            if exc is None:
                # 协程初始化（生成器或者协程初始化 next(xxx））
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            if self._must_cancel:
                # 在中止以前取消任务
                self._must_cancel = False
                super().set_exception(futures.CancelledError())
            else:
                # 拿到了协程/生成器的结果
                super().set_result(exc.value)
        except futures.CancelledError:
            super().cancel()  # I.e., Future.cancel(self).
        except Exception as exc:
            super().set_exception(exc)
        except BaseException as exc:
            super().set_exception(exc)
            raise
    ...

import asyncio

async def handler(client_reader, client_writer):
    # 没有数据就阻塞等（主线程作其余事情去了）
    data = await client_reader.read(2048)
    print(data.decode("utf-8"))

    client_writer.write("骊山语罢清宵半,泪雨霖铃终不怨\n何如薄幸锦衣郎,比翼连枝当日愿".encode("utf-8"))
    await client_writer.drain()  # 等待缓冲区（缓冲区没占满就直接返回）
    client_writer.close()  # 关闭链接

async def main():
    server = await asyncio.start_server(handler, "127.0.0.1", 8080)
    print("Server已经启动，端口：8080")
    # 实现了协程方法`__aenter__`和`__aexit__`的可使用`async with`
    async with server:
        # async def serve_forever(self):pass ==> use await
        await server.serve_forever()  # 异步方法

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def main():
    reader, writer = await asyncio.open_connection("127.0.0.1", 8080)
    writer.write("人生若只如初见,何事秋风悲画扇\n等闲变却故人心,却道故人心易变".encode("utf-8"))
    data = await reader.read(2048)
    if data:
        print(data.decode("utf-8"))

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)

import asyncio

async def get_html(host):
    print("get_html %s..." % host)
    reader, writer = await asyncio.open_connection(host, 80)
    writer.write(f"GET / HTTP/1.1\r\nHost: {host}\r\n\r\n".encode('utf-8'))
    await writer.drain()  # 等待缓冲区

    html_list = []
    async for line in reader:
        html_list.append(line.decode("utf-8"))

    writer.close()  # 关闭链接
    return "\n".join(html_list)

async def main():
    tasks = [
        asyncio.create_task(get_html(url))
        for url in ['dotnetcrazy.cnblogs.com', 'dunitian.cnblogs.com']
    ]
    html_list = await asyncio.gather(*tasks)
    print(html_list)

if __name__ == "__main__":
    import time
    start_time = time.time()

    asyncio.run(main())

    print(time.time() - start_time)