Python 异步编程

def fun():
    print('---------start---------')
    while True:
        yield '生成器返回的数据....'


g = fun()
print(next(g))
print('*' * 20)
print(next(g))

import time


def func_a():
    while True:
        print('-------A-------')
        yield
        time.sleep(0.5)


def func_b(obj):
    while True:
        print('---------B------------')
        obj.__next__()


a = func_a()
func_b(a)

import time


def foo():
    time.sleep(1)


now = lambda: time.time()
start = now()

for i in range(5):
    foo()

print('同步所花费的时间: %f s' % (now() - start))

import time
import asyncio

now = lambda: time.time()

async def fun():
    await asyncio.sleep(1)  # 正确的异步写法
    print('我是async任务...')

async def main():
    tasks = [fun() for _ in range(5)]  # 创建多个异步任务
    await asyncio.gather(*tasks)  # 并发执行所有任务

start = now()	

asyncio.run(main())  # 启动主事件循环

print('异步所花费的时间: %f s' % (now() - start))

import asyncio

# 异步任务1
async def work_1():
    for _ in range(5):
        print('我是异步任务1')
        await asyncio.sleep(1)

# 异步任务2
async def work_2():
    for _ in range(5):
        print('我是异步任务2')
        await asyncio.sleep(1)

tasks = [
    work_1(),
    work_2()
]

async def main():
    await asyncio.gather(*tasks)

asyncio.run(main())

import asyncio

# 定义一个异步函数
async def my_coroutine():
    print("Hello from the coroutine!")
    await asyncio.sleep(1)
    print("Goodbye from the coroutine!")

# 定义一个异步的 main 函数，直接运行协程
async def main():
    await my_coroutine()

# 运行 main 协程
asyncio.run(main())

# 定义一个异步的 main 函数，使用 create_task 创建任务
async def main():
    task = asyncio.create_task(my_coroutine())  # 将协程包装为任务
    await task  # 等待任务完成

# 运行 main 协程
asyncio.run(main())

import asyncio

# 定义异步任务
async def task_1():
    print("Task 1 starting...")
    await asyncio.sleep(1)
    print("Task 1 done.")

async def task_2():
    print("Task 2 starting...")
    await asyncio.sleep(1)
    print("Task 2 done.")

# 使用 asyncio.run 和 asyncio.gather 同时运行多个任务
async def main():
    await asyncio.gather(task_1(), task_2())

asyncio.run(main())

import asyncio
from requests import get


def send_request():
    response = get("http://localhost:5000")
    return f"接收到响应：{response.text}"

async def main():
    response = await send_request()
    print(response)

asyncio.run(main())

import asyncio

async def others():
    print('start')
    await asyncio.sleep(2)  # 当前等待的sleep对象的返回值为None
    print('end')
    return '返回值'

async def func():
    print('执行协程函数内部代码')
    # 遇到阻塞操作，挂起该协程。当阻塞操作完成后再继续往下执行
    # 当前协程挂起时，事件循环可以执行其他协程
    response = await others()  # others()是一个协程对象
    print('结果:', response)

asyncio.run(func())

import asyncio

async def others():
    print('start')
    await asyncio.sleep(2)
    print('end')
    return '返回值'

async def func():
    print('执行协程函数内部代码')
    response_1 = await others()
    print('response_1的IO操作结果为:', response_1)

    response_2 = await others()
    print('response_2的IO操作结果为:', response_2)

asyncio.run(func())

import asyncio

async def others():
    print('start')
    await asyncio.sleep(2)
    print('end')
    return '返回值'

async def func():
    print('执行协程函数内部代码')

    # python3.7以上版本写法
    task_1 = asyncio.create_task(others())
    task_2 = asyncio.create_task(others())

    response_1 = await task_1
    print('task_1的IO操作结果为:', response_1)
    response_2 = await task_2
    print('task_2的IO操作结果为:', response_2)

asyncio.run(func())

import asyncio

async def others(time, name):
    print('start', name)
    await asyncio.sleep(time)
    print('end', name)
    return '返回值' + name

async def func():
    print('执行协程函数内部代码')

    task_1 = asyncio.create_task(others(3, 'task_1'))
    task_2 = asyncio.create_task(others(2, 'task_2'))

    response_1 = await task_1
    print('task_1的IO操作结果为:', response_1)
    response_2 = await task_2
    print('task_2的IO操作结果为:', response_2)

asyncio.run(func())

import asyncio

async def others(time, name):
    print('start', name)
    await asyncio.sleep(time)
    print('end', name)
    return '返回值' + name

async def func():
    print('执行协程函数内部代码')

    task_1 = asyncio.create_task(others(2, 'task_1'))
    task_2 = asyncio.create_task(others(3, 'task_2'))

    response_1 = await task_1
    print('task_1的IO操作结果为:', response_1)
    response_2 = await task_2
    print('task_2的IO操作结果为:', response_2)

asyncio.run(func())

import time
import asyncio


# 使用async 来定义一个协程对象
async def do_work(x):
    print('waiting:', x)
    await asyncio.sleep(x)  # 模拟一个IO操作
    return 'Done after {}s'.format(x)


async def main():
    # 创建多个协程对象
    coroutine_1 = do_work(1)
    coroutine_2 = do_work(2)
    coroutine_3 = do_work(3)

    tasks = [# 在列表中创建多个task对象
        asyncio.create_task(coroutine_1), asyncio.create_task(coroutine_2),
        asyncio.create_task(coroutine_3)]

    # 使用 asyncio.wait 来等待任务完成，timeout=None 表示等待直到所有任务完成
    dones, pendings = await asyncio.wait(tasks, timeout=None)

    for task in dones:
        print('task result:', task.result())

    print("pendings:", pendings)


now = lambda: time.time()
start = now()

# 使用 asyncio.run 来启动协程和事件循环
asyncio.run(main())

print('Time:', now() - start)

coroutine_1 = do_work(1)
coroutine_2 = do_work(2)
coroutine_3 = do_work(3)

tasks = [
   asyncio.create_task(coroutine_1, name='MY_COROUTINE_1'),
   asyncio.create_task(coroutine_2, name='MY_COROUTINE_2'),
   asyncio.create_task(coroutine_3, name='MY_COROUTINE_3')
]

import asyncio

async def main():
    # 创建一个任务（future对象），什么也不干
    fut = asyncio.Future()  # 创建一个 Future 对象
    # await会等待任务最终结果，没有结果则会一直等待下去
    await fut

asyncio.run(main())

import asyncio

async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result('666')

async def main():
    # 创建一个任务（future对象），什么也不干
    fut = asyncio.Future()  # 创建 Future 对象

    # 等待任务最终结果，没有结果则会一直等待下去
    # 手动设置future的最终结果，那样fut就可以结束了
    await asyncio.create_task(set_after(fut))  # 使用 create_task 来创建任务并调度执行
    data = await fut  # 等待future对象获取最终结果，否则一直等待下去
    print(data)

# 使用 asyncio.run() 来执行协程，避免手动管理事件循环
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)
    return 123


# 创建线程池
pool = ThreadPoolExecutor(max_workers=2)
# 或者创建进程池
# pool =  ProcessPoolExecutor(max_workers=5)

for i in range(4):
    fut = pool.submit(func, i)  # 返回的是一个Future对象, fut.result()才是返回结果
    print(fut)
    # print(fut.result())

import time
import asyncio
import concurrent.futures

def func_1():
    time.sleep(2)
    return '这是一个任务的返回值...'

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

    # 在协程函数中运行普通函数 在执行函数时，协程内部会自动创建一个线程池来运行任务
    # run_in_executor()方法第一个参数为None时则默认创建一个线程池
    fut = loop.run_in_executor(None, func_1)
    result = await fut
    print('当前方式会自动创建一个线程池去执行普通函数: ', result)

    # 在协程函数中运行基于线程池的任务, 效果与以上代码一致
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, func_1)
        print('在线程池中得到的执行结果: ', result)

    # 在协程函数中运行基于进程池的任务
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, func_1)
        print('在进程池中得到的执行结果: ', result)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import requests

# 异步下载图片函数
async def download_image(url):
    # 发送网络请求下载图片
    print('开始下载:', url)
    # request不支持异步，使用线程池来辅助实现
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(None, requests.get, url)
    response = await future
    print('下载完成')
    
    # 图片保存
    file_name = url.rsplit('/')[-1]
    with open(file_name, 'wb') as f:
        f.write(response.content)

# 主程序
async def main():
    url_list = [
        'http://pic.bizhi360.com/bbpic/98/10798.jpg',
        'http://pic.bizhi360.com/bbpic/92/10792.jpg',
        'http://pic.bizhi360.com/bbpic/86/10386.jpg'
    ]
    
    # 使用 asyncio.create_task 来创建协程任务
    tasks = [asyncio.create_task(download_image(url)) for url in url_list]

    # 等待所有任务完成
    await asyncio.gather(*tasks)

# 使用 asyncio.run 来启动整个程序
if __name__ == '__main__':
    asyncio.run(main())

import asyncio

class AsyncCounter:
    def __init__(self, start, end):
        self.current = start
        self.end = end

    # 返回迭代器本身
    def __aiter__(self):
        return self
    
    # 获取下一个值，异步获取
    async def __anext__(self):
        if self.current > self.end:
            raise StopAsyncIteration  # 结束迭代
        await asyncio.sleep(1)  # 模拟异步操作（比如从数据库中读取）
        self.current += 1
        return self.current - 1

# 异步遍历数据
async def main():
    async for number in AsyncCounter(1, 5):
        print(f"Received number: {number}")

# 运行异步任务
asyncio.run(main())

import asyncio

class AsyncFileReader:
    def __init__(self, filename):
        self.filename = filename

    def __aiter__(self):
        self.file = open(self.filename, 'r')
        return self
    
    async def __anext__(self):
        line = await asyncio.to_thread(self.file.readline)
        if not line:
            self.file.close()
            raise StopAsyncIteration
        return line.strip()

async def main():
    async for line in AsyncFileReader('sample.txt'):
        print(f"读取到的行：{line}")

asyncio.run(main())

import asyncio

# 使用async 来定义一个协程对象
async def do_work(x):
    print('waiting:', x)
    await asyncio.sleep(x)
    return 'Done after {} s'.format(x)

# 给任务添加回调函数
def callback(future):
    print('callback:', future.result())

async def run_task():
    # 获取协程对象
    coroutine = do_work(3)
    # 创建task
    task = asyncio.create_task(coroutine)
    # 给task添加回调
    task.add_done_callback(callback)
    return await task

# 将协程对象加入到事件循环中
asyncio.run(run_task())  # 注意这里加上了括号，调用异步任务

import asyncio

# 使用async 来定义一个协程对象
async def do_work(x):
    print('waiting:', x)
    await asyncio.sleep(x)
    return f'Done after {x} s'

# 获取协程对象
async def main():
    coroutine = do_work(3)
    # 创建task
    task = asyncio.create_task(coroutine)

    # 等待任务完成
    result = await task
    print('直接获取返回结果:', result)

# 使用asyncio.run来启动事件循环
asyncio.run(main())

import time
import asyncio

# 使用async 来定义一个协程对象
async def do_work(x):
    print('waiting:', x)
    await asyncio.sleep(x)  # 模拟一个IO操作
    return f'Done after {x} s'

async def main():
    now = lambda: time.time()
    start = now()
    # 获取协程对象
    coroutine = do_work(3)
    # 创建task
    task = asyncio.create_task(coroutine)
    # 将协程对象加入到事件循环中
    await task
    # 下面的程序会阻塞住，直到协程执行完毕
    print('task result:', task.result())
    print('Time:', now() - start)

asyncio.run(main())

import time
import asyncio

now = lambda: time.time()


async def func():
    task_1 = asyncio.ensure_future(asyncio.sleep(1))
    task_2 = asyncio.ensure_future(asyncio.sleep(1))
    await task_1
    await task_2


async def main():
    start = now()
    for i in range(5):
        await func()
    print('异步所花费的时间: %f s' % (now() - start))

asyncio.run(main())

import time
import asyncio

now = lambda: time.time()


async def func():
    await asyncio.ensure_future(asyncio.sleep(1))
    await asyncio.ensure_future(asyncio.sleep(1))


async def main():
    start = now()
    for i in range(5):
        await func()
    print('异步所花费的时间: %f s' % (now() - start))

asyncio.run(main())

import asyncio

class AsyncContextManager:
    async def do_something(self):
        return '模拟业务操作'

    async def __aenter__(self):
        print("进入上下文管理器")
        await asyncio.sleep(1)
        return self

    async def __aexit__(self, exec_type, exc_val, exc_tb):
        """
         with语句运行结束之后触发此方法的运行
         exc_type:如果抛出异常, 这里获取异常类型
         exc_val:如果抛出异常, 这里显示异常内容
         exc_tb:如果抛出异常, 这里显示所在位置, traceback
         """
        print("退出上下文管理器")
        await asyncio.sleep(1)

async def main():
    async with AsyncContextManager() as fp:
        result = await fp.do_something()
        print(result)

asyncio.run(main())

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    url = "http://example.com"
    html = await fetch(url)
    print(html)

asyncio.run(main())

# pip install aiomysql

import asyncio
import aiomysql

async def execute():
    # 网络IO操作，连接mysql
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='root', db='test')
    # 网络IO操作，创建cursor()
    cur = await conn.cursor()
    # 网络IO操作，执行sql语句
    await cur.execute('SELECT host.user FROM user')
    # 网络IO操作，获取sql结果
    result = cur.fetchall()
    print(result)
    # 网络IO操作，关闭链接
    await cur.close()
    conn.close()

asyncio.run(execute())

import asyncio
import aiomysql

async def execute(host, password):
    # 网络IO操作，连接mysql
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='test')
    # 网络IO操作，创建cursor()
    cur = await conn.cursor()
    # 网络IO操作，执行sql语句
    await cur.execute('SELECT host.user FROM user')
    # 网络IO操作，获取sql结果
    result = cur.fetchall()
    print(result)
    # 网络IO操作，关闭链接
    await cur.close()
    conn.close()

async def main():
    task_list = [
        asyncio.ensure_future(execute('127.0.0.1', 'root')),
        asyncio.ensure_future(execute('40.0.0.1', 'root'))
    ]

    await asyncio.wait(task_list)

asyncio.run(execute())