大数据

python yield于asyncio使用

Posted by

yield

使用方法

demo01

# yield 像是一个可暂停的return 下次next()从暂停处继续
def my_first_gen():
    print(">>> 启动生成器")
    yield 100
    print(">>> 第一次修复")
    yield 200
    print(">>> 第二次修复")
    yield 300
    print(">>> 生成器彻底结束")

# 注意这里不会打印
g = my_first_gen()
print(f"生成器对象：{g}")

# 逐次驱动
print(next(g))
print(next(g))
print(next(g))
# print(next(g)) # StopIteration

demo02

def my_range(start, end):
    current = start
    while current < end:
        yield current  # 每循环一次 吐出一个数字
        current += 1


for num in my_range(0, 5):
    print(f"当前数字: {num}")

# 验证内存占用（对比列表）
import sys

list_data = [i for i in range(1000000)]
gen_data = my_range(0, 1000000)
print(f"列表内存：{sys.getsizeof(list_data)} 字节")
print(f"生成器内存：{sys.getsizeof(gen_data)} 字节")

# 只要你想流式处理数据（渡日治、分页拉取API） 第一时间想 yield

demo03

def calc_coroutine():
    print("计算器已启动，等待输入...")
    total = 0
    while True:
        # 1. 暂停在这里，等待外部send值进来
        # 2. 外部 send(10) 会把10 赋值给cmd
        cmd = yield total  # yield total将total值返回给调用者 后面 .send()会将send的值赋值给cmd

        if cmd == "reset":
            total = 0
        elif isinstance(cmd, (int, float)):
            total += cmd
        # 如果外部调用close()循环结束

c = calc_coroutine()
# 必须首次执行一个next() 让程序运行到第一个yield处（即等待输入状态）
print(next(c))
print(c.send(10))
print(c.send(5))
print(c.send("reset"))
c.close()

demo04

# yield from 子生成器 当甩手掌柜 我不干了，让子生成器直接跟调用者对话，我是透明的。

def flatten(nested_list):
    for sublist in nested_list:
        # 如果子元素还是列表 递归咱开
        if isinstance(sublist, list):
            # yield from 是直通车 / 传话筒。我不干活，让给子生成器干；
            # 子生成器吐出来的所有东西，我原封不动递给调用者。
            # 遇到子列表，我就原地立正。子生成器生产的所有数据，不经过我手，直接‘瞬移’到调用者面前。
            yield from flatten(sublist)  # 把子生成器的值透传
        else:
            yield sublist


data = [1, [2, 3], [[4, 5], 6]]
for item in flatten(data):
    print(item, end=" ")

asyncio

demo05

# async def 定义协程 await 挂起当前协程去执行别的任务 asyncio.run() 启动事件循环来调度所有任务

import time
import asyncio


# 同步写法 耗时3秒
def sync_task():
    time.sleep(1)  # 死等


start = time.time()
for _ in range(3):
    sync_task()
print(f"同步耗时：{time.time() - start:.2f} 秒")


# 异步写法
async def async_task():
    await asyncio.sleep(1)  # 假装等，实际让出控制权


async def main():
    # 创建3个任务 并发执行
    tasks = [async_task() for _ in range(3)]
    await asyncio.gather(*tasks)


start = time.time()
asyncio.run(main())
print(f"异步并发耗时：{time.time() - start:.2f} 秒")

demo06

import asyncio

# 协程不 await，就像汽车不发动
async def say_hello():
    print("Hello")
    return "World"


async def main():
    # 错误：直接调到用say_hello()返回的是协程对象，不会执行打印
    coro = say_hello()
    print(f"类型：{coro}")

    # 正确: 必须await 才能执行
    result = await say_hello()
    print(f"返回值：{result}")


asyncio.run(main())

demo07

import asyncio


async def worker(name, delay, fail=False):
    print(f"Worker {name} 启动，等待 {delay}s")
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"Worker {name} 爆炸了")
    return f"Worker {name} 的结果"


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(worker("A", 1))
            t2 = tg.create_task(worker("B", 0.5, fail=True))  # 这个会报错
            t3 = tg.create_task(worker("C", 2))
        # 如果t2报错， t1和t3会被自动取消，不会默认后台运行
        print([t.result() for t in [t1, t2, t3]])
    except ValueError as e:
        print(f"捕获到任务组异常：{e}")


asyncio.run(main())

demo08

import asyncio


async def log_stream():
    for i in range(5):
        # 模拟从远程kafka或websocket拉取数据，耗时1秒
        await asyncio.sleep(1)
        yield f"日志条目 #{i} 时间：{asyncio.get_running_loop().time():.2f}"


# async for 专门用来遍历异步生成器 每次循环遇到await会让出控制权，非常适合流式数据接口

async def consumer():
    # 使用async for 迭代异步生成器
    async for log in log_stream():
        print(f"处理：{log}")
        # 可以在这里调用另一个异步函数保存到数据库


asyncio.run(consumer())

demo09

import asyncio
import random


# 模拟异步HTTP请求
async def mock_api_request(url_id):
    delay = random.uniform(0.5, 1.5)  # 模拟网络延迟
    await asyncio.sleep(delay)
    return f"Response from API-{url_id} (耗时 {delay:.2f})s"


# 异步生成器：产生并发请求的结果（按完成顺序）
async def fetch_all(url_ids):
    async with asyncio.TaskGroup() as tg:
        # 创建所有请求任务
        tasks = {tg.create_task(mock_api_request(id)): id for id in url_ids}

        # 核心技巧：等待每个任务完成，完成后 yield 结果
        # 因为 TaskGroup 不能直接 yield，我们配合 asyncio.as_completed 来做
        # 但 as_completed 不适用 TaskGroup 内部。改进写法：
        pass  # 为了教学清晰，我们改用更直接的 as_completed（不依赖 TaskGroup）


# 更推荐的纯净写法（放弃 TaskGroup 的强制绑定，使用 as_completed）
async def fetch_all_v2(url_ids):
    tasks = [asyncio.create_task(mock_api_request(id)) for id in url_ids]
    # as_completed 返回迭代器，每完成一个任务就产出
    for coro in asyncio.as_completed(tasks):
        result = await coro
        yield result


async def main():
    print("开始并发抓取...")
    # 异步迭代，只要任何一个 API 返回，立刻打印，不用等全部完成
    async for data in fetch_all_v2([1, 2, 3, 4, 5]):
        print(f">>> 收到数据：{data}")
    print("所有数据接受完毕！")


if __name__ == '__main__':
    asyncio.run(main())
下一篇
shell 知识点回顾
上一篇 python 正则知识点回顾