任務(wù)列表 = [任務(wù)1， 任務(wù)2， 任務(wù)3............]
while True:
    可執(zhí)行的任務(wù)列表，已完成的任務(wù)列表 = 去任務(wù)列表中檢查所有的任務(wù)，將'可執(zhí)行'和'已完成'的任務(wù)返回
    for 就緒任務(wù) in 可執(zhí)行的任務(wù)列表：
        執(zhí)行已就緒的任務(wù)
    for 已完成的任務(wù) in 已完成的任務(wù)列表：
        在任務(wù)列表中移除 已完成的任務(wù)
    如果任務(wù)列表中的任務(wù)都已完成，則終止循環(huán)

import asyncio
# 獲取事件循環(huán)
loop = asyncio.get_event_loop()
# 將任務(wù)放到`任務(wù)列表`，監(jiān)聽(tīng)事件循環(huán)
loop.run_until_complete(任務(wù))
# 關(guān)閉事件
loop.close()

# 定義協(xié)程函數(shù)
async def func():
    pass
# 創(chuàng)建協(xié)程對(duì)象
result = func()

import asyncio
async def func():
    print("你好")
result = func()
# 方式1
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
# 方式2
asyncio.run(result)  # python3.7寫(xiě)法

import asyncio
async def func():
    print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")
    # 遇到IO操作掛起當(dāng)前協(xié)程（任務(wù)），等IO操作完成之后再繼續(xù)往下執(zhí)行。
    # 當(dāng)前協(xié)程掛起時(shí)，事件循環(huán)可以去執(zhí)行其他協(xié)程（任務(wù)）。
    response = await asyncio.sleep(2)
    print("IO請(qǐng)求結(jié)束，結(jié)果為：", response)
result = func()
asyncio.run(result)

import asyncio
async def others():
    print("start")  # ④打印start
    await asyncio.sleep(2)  # ⑤等待耗時(shí)2秒，這過(guò)程中可以切換到其他協(xié)程中去
    print("end")  # ⑥打印end
    return '返回值'
async def func():
    print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")  # ②執(zhí)行協(xié)程函數(shù)，打印print代碼
    response = await others()  # ③等待協(xié)程函數(shù)others
    print(f"io請(qǐng)求結(jié)束，結(jié)果為{response}")  # ⑦等待others結(jié)束后打印print語(yǔ)句
if __name__ == '__main__':
    asyncio.run(func())  # ①協(xié)程函數(shù)放入事件循環(huán)中運(yùn)行

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main開(kāi)始")
    # 創(chuàng)建協(xié)程，將協(xié)程封裝到一個(gè)Task對(duì)象中并立即添加到事件循環(huán)的任務(wù)列表中，等待事件循環(huán)去執(zhí)行（默認(rèn)是就緒狀態(tài)）。
    task1 = asyncio.create_task(func())
    # 創(chuàng)建協(xié)程，將協(xié)程封裝到一個(gè)Task對(duì)象中并立即添加到事件循環(huán)的任務(wù)列表中，等待事件循環(huán)去執(zhí)行（默認(rèn)是就緒狀態(tài)）。
    task2 = asyncio.create_task(func())
    print("main結(jié)束")
    # 當(dāng)執(zhí)行某協(xié)程遇到IO操作時(shí)，會(huì)自動(dòng)化切換執(zhí)行其他任務(wù)。
    # 此處的await是等待相對(duì)應(yīng)的協(xié)程全都執(zhí)行完畢并獲取結(jié)果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)
asyncio.run(main())

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main開(kāi)始")
    # 創(chuàng)建協(xié)程，將協(xié)程封裝到Task對(duì)象中并添加到事件循環(huán)的任務(wù)列表中，等待事件循環(huán)去執(zhí)行（默認(rèn)是就緒狀態(tài)）。
    # 在調(diào)用
    task_list = [asyncio.create_task(func()), asyncio.create_task(func())]
    print("main結(jié)束")
    # 當(dāng)執(zhí)行某協(xié)程遇到IO操作時(shí)，會(huì)自動(dòng)化切換執(zhí)行其他任務(wù)。
    # 此處的await是等待所有協(xié)程執(zhí)行完畢，并將所有協(xié)程的返回值保存到done
    # 如果設(shè)置了timeout值，則意味著此處最多等待的秒，完成的協(xié)程返回值寫(xiě)入到done中，未完成則寫(xiě)到pending中。
    done, pending = await asyncio.wait(task_list)
    print(done)
asyncio.run(main())

async def main():
    # 獲取當(dāng)前事件循環(huán)
    loop = asyncio.get_running_loop()
    # 創(chuàng)建一個(gè)任務(wù)（Future對(duì)象），這個(gè)任務(wù)什么都不干。
    fut = loop.create_future()
    # 等待任務(wù)最終結(jié)果（Future對(duì)象），沒(méi)有結(jié)果則會(huì)一直等下去。
    await fut
asyncio.run(main())

import asyncio
async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("666")
async def main():
    # 獲取當(dāng)前事件循環(huán)
    loop = asyncio.get_running_loop()
    # 創(chuàng)建一個(gè)任務(wù)（Future對(duì)象），沒(méi)綁定任何行為，則這個(gè)任務(wù)永遠(yuǎn)不知道什么時(shí)候結(jié)束。
    fut = loop.create_future()
    # 創(chuàng)建一個(gè)任務(wù)（Task對(duì)象），綁定了set_after函數(shù)，函數(shù)內(nèi)部在2s之后，會(huì)給fut賦值。
    # 即手動(dòng)設(shè)置future任務(wù)的最終結(jié)果，那么fut就可以結(jié)束了。
    await loop.create_task(set_after(fut))
    # 等待 Future對(duì)象獲取 最終結(jié)果，否則一直等下去
    data = await fut
    print(data)
asyncio.run(main())

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
def func(value):
    time.sleep(1)
    print(value)
pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)
for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

import time
import asyncio
import concurrent.futures
def func1():
    # 某個(gè)耗時(shí)操作
    time.sleep(2)
    return "OK"
async def main():
    loop = asyncio.get_running_loop()
    # 方式1. Run in the default loop's executor ( 默認(rèn)ThreadPoolExecutor )
    # 第一步：內(nèi)部會(huì)先調(diào)用 ThreadPoolExecutor 的 submit 方法去線程池中申請(qǐng)一個(gè)線程去執(zhí)行func1函數(shù)，并返回一個(gè)concurrent.futures.Future對(duì)象
    # 第二步：調(diào)用asyncio.wrap_future將concurrent.futures.Future對(duì)象包裝為asycio.Future對(duì)象。
    # 因?yàn)閏oncurrent.futures.Future對(duì)象不支持await語(yǔ)法，所以需要包裝為 asycio.Future對(duì)象 才能使用。
    fut = loop.run_in_executor(None, func1)
    result = await fut
    print('default thread pool', result)
    # 方式2. Run in a custom thread pool:
    # with concurrent.futures.ThreadPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom thread pool', result)
    # 方式3. Run in a custom process pool:
    # with concurrent.futures.ProcessPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom process pool', result)
asyncio.run(main())

import asyncio
import requests
async def download_image(url):
    # 發(fā)送網(wǎng)絡(luò)請(qǐng)求，下載圖片（遇到網(wǎng)絡(luò)下載圖片的IO請(qǐng)求，自動(dòng)化切換到其他任務(wù)）
    print("開(kāi)始下載:", url)
    loop = asyncio.get_event_loop()
    # requests模塊默認(rèn)不支持異步操作，所以就使用線程池來(lái)配合實(shí)現(xiàn)了。
    future = loop.run_in_executor(None, requests.get, url)
    response = await future
    print('下載完成')
    # 圖片保存到本地文件
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)
if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]
    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete( asyncio.wait(tasks) )

import asyncio
class Reader(object):
    """ 自定義異步迭代器（同時(shí)也是異步可迭代對(duì)象） """
    def __init__(self):
        self.count = 0
    async def readline(self):
        # await asyncio.sleep(1)
        self.count += 1
        if self.count == 100:
            return None
        return self.count
    def __aiter__(self):
        return self
    async def __anext__(self):
        val = await self.readline()
        if val == None:
            raise StopAsyncIteration
        return val
async def func():
    # 創(chuàng)建異步可迭代對(duì)象
    async_iter = Reader()
    # async for 必須要放在async def函數(shù)內(nèi)，否則語(yǔ)法錯(cuò)誤。
    async for item in async_iter:
        print(item)
asyncio.run(func())

import asyncio
class AsyncContextManager:
    def __init__(self):
        self.conn = None
    async def do_something(self):
        # 異步操作數(shù)據(jù)庫(kù)
        return 666
    async def __aenter__(self):
        # 異步鏈接數(shù)據(jù)庫(kù)
        self.conn = await asyncio.sleep(1)
        return self
    async def __aexit__(self, exc_type, exc, tb):
        # 異步關(guān)閉數(shù)據(jù)庫(kù)鏈接
        await asyncio.sleep(1)
async def func():
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)
asyncio.run(func())

pip3 install uvloop

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 編寫(xiě)asyncio的代碼，與之前寫(xiě)的代碼一致。
# 內(nèi)部的事件循環(huán)自動(dòng)化會(huì)變?yōu)閡vloop
asyncio.run(...)

pip3 install aioredis

import asyncio
import aioredis
async def execute(address, password):
    print("開(kāi)始執(zhí)行", address)
    # 網(wǎng)絡(luò)IO操作：先去連接 77.95.4.197:6379，遇到IO則自動(dòng)切換任務(wù)，去連接77.95.4.198:6379
    redis = await aioredis.create_redis_pool(address, password=password)
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    redis.close()
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    await redis.wait_closed()
    print("結(jié)束", address)
task_list = [
    execute('redis://77.95.4.197:6379', "123456"),
    execute('redis://77.95.4.198:6379', "123456")
]
asyncio.run(asyncio.wait(task_list))

pip3 install aiomysql

import asyncio
import aiomysql
async def execute(host, password):
    print("開(kāi)始", host)
    # 網(wǎng)絡(luò)IO操作：先去連接 77.95.40.197，遇到IO則自動(dòng)切換任務(wù)，去連接77.95.40.198:6379
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    cur = await conn.cursor()
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    await cur.execute("SELECT Host,User FROM user")
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    result = await cur.fetchall()
    print(result)
    # 網(wǎng)絡(luò)IO操作：遇到IO會(huì)自動(dòng)切換任務(wù)
    await cur.close()
    conn.close()
    print("結(jié)束", host)
task_list = [
    execute('77.95.40.197', "123456"),
    execute('77.95.40.198', "123456")
]
asyncio.run(asyncio.wait(task_list))

pip3 install aiohttp

import aiohttp
import asyncio
async def fetch(session, url):
    print(f"發(fā)送請(qǐng)求：{url}")
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print("得到結(jié)果：", url, len(text))
async def main():
    async with aiohttp.ClientSession() as session:
        url_list = ["http://www.baidu.com", "http://www.taobao.com", "http://www.jd.com"]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
if __name__ == '__main__':
    asyncio.run(main())