事件循环与协程深入

import asyncio
import time

async def task(name: str, delay: float) -> str:
    print(f"[{time.strftime('%H:%M:%S')}] {name} 开始执行")
    await asyncio.sleep(delay)  # 模拟 IO 操作
    print(f"[{time.strftime('%H:%M:%S')}] {name} 执行完毕")
    return f"{name} 的结果"

async def main():
    # 创建多个 Task，观察事件循环如何交替调度
    t1 = asyncio.create_task(task("任务A", 2))
    t2 = asyncio.create_task(task("任务B", 1))
    t3 = asyncio.create_task(task("任务C", 3))

    # 事件循环会在 Task 挂起时切换执行其他 Task
    results = await asyncio.gather(t1, t2, t3)
    print(f"所有结果: {results}")

asyncio.run(main())

[14:30:01] 任务A 开始执行
[14:30:01] 任务B 开始执行
[14:30:01] 任务C 开始执行
[14:30:02] 任务B 执行完毕    # 1秒后最先完成
[14:30:03] 任务A 执行完毕    # 2秒后完成
[14:30:04] 任务C 执行完毕    # 3秒后完成
所有结果: ['任务A 的结果', '任务B 的结果', '任务C 的结果']

# asyncio.runners 模块简化实现
def run(main, *, debug=None):
    # 1. 安全校验：确保没有已运行的事件循环
    if events._get_running_loop() is not None:
        raise RuntimeError("asyncio.run() cannot be called from a running event loop")

    # 2. 确保传入的是协程
    if not coroutines.iscoroutine(main):
        raise ValueError("a coroutine was expected, got {!r}".format(main))

    # 3. 创建新事件循环
    loop = events.new_event_loop()

    try:
        events.set_event_loop(loop)
        if debug is not None:
            loop.set_debug(debug)
        # 4. 运行主协程
        return loop.run_until_complete(main)
    finally:
        try:
            # 5. 取消所有未完成的 Task
            _cancel_all_tasks(loop)
            # 6. 运行直到所有 Task 彻底结束
            loop.run_until_complete(loop.shutdown_asyncgens())
            # Python 3.9+: 关闭延迟回调
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            # 7. 关闭事件循环
            events.set_event_loop(None)
            loop.close()

import asyncio

# 方式1: 获取当前线程的事件循环（如果没有则创建）
loop = asyncio.get_event_loop()

# 方式2: 获取当前正在运行的事件循环（如果没有则抛出异常）
try:
    loop = asyncio.get_running_loop()
except RuntimeError:
    print("没有正在运行的事件循环")

# 方式3: 创建新的事件循环（不设置为当前循环）
loop = asyncio.new_event_loop()

# 方式4: 创建并设置为当前线程的事件循环
loop = asyncio.set_event_loop(asyncio.new_event_loop())

import asyncio

async def fetch_data() -> str:
    await asyncio.sleep(1)
    return "数据获取完成"

# 手动管理事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    result = loop.run_until_complete(fetch_data())
    print(result)  # 数据获取完成
finally:
    loop.close()

import asyncio

async def periodic_task():
    """每2秒执行一次的周期任务"""
    count = 0
    while True:
        count += 1
        print(f"周期任务第 {count} 次执行")
        await asyncio.sleep(2)

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# 启动周期任务
task = loop.create_task(periodic_task())

# 5秒后停止循环
loop.call_later(5, loop.stop)

try:
    loop.run_forever()
finally:
    task.cancel()  # 取消未完成的任务
    loop.run_until_complete(task)  # 等待取消完成
    loop.close()

import asyncio

async def async_operation() -> str:
    await asyncio.sleep(1)
    return "异步操作完成"

def sync_caller() -> str:
    """从同步代码中调用异步函数的正确方式"""
    try:
        # 优先尝试获取已运行的循环（如在 Jupyter 中）
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop and loop.is_running():
        # 已有运行中的循环，使用 nest_asyncio 或线程方案
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor() as pool:
            future = pool.submit(asyncio.run, async_operation())
            return future.result()
    else:
        # 没有运行中的循环，直接创建
        return asyncio.run(async_operation())

result = sync_caller()
print(result)  # 异步操作完成

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "完成"

# 协程 vs Task 的区别
coro = my_coroutine()       # 协程对象，尚未执行
task = asyncio.create_task(my_coroutine())  # Task，已被调度到事件循环

import asyncio

async def work(name: str) -> str:
    await asyncio.sleep(1)
    return f"{name} 完成"

async def main():
    # 方式1: asyncio.create_task() — 推荐（Python 3.7+）
    task1 = asyncio.create_task(work("任务1"), name="my-task-1")

    # 方式2: loop.create_task() — 底层 API
    loop = asyncio.get_running_loop()
    task2 = loop.create_task(work("任务2"))

    # 方式3: asyncio.ensure_future() — 兼容旧代码
    task3 = asyncio.ensure_future(work("任务3"))

    # 方式4: Python 3.11+ TaskGroup — 结构化并发
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(work("TG任务1"))
        t2 = tg.create_task(work("TG任务2"))
    # TaskGroup 退出时自动等待所有 Task 完成

    results = await asyncio.gather(task1, task2, task3)
    print(results)

import asyncio

async def long_running_task():
    try:
        print("任务开始")
        await asyncio.sleep(10)  # 模拟长时间 IO
        print("任务完成")  # 这行不会执行
    except asyncio.CancelledError:
        print("任务被取消，执行清理")
        raise  # 重要：重新抛出 CancelledError
    finally:
        print("清理资源")

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

    # 等待一小段时间后取消
    await asyncio.sleep(1)
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print("主协程感知到任务被取消")

asyncio.run(main())

任务开始
任务被取消，执行清理
清理资源
主协程感知到任务被取消

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(1)
    if "error" in url:
        raise ValueError(f"获取 {url} 失败")
    return f"数据来自 {url}"

async def main():
    urls = ["http://api1.com", "http://error.com", "http://api2.com"]

    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(url)) for url in urls]
    except* ValueError as eg:
        # ExceptionGroup 语法：只处理 ValueError 类型的异常
        print(f"部分任务失败: {eg.exceptions}")
    else:
        results = [t.result() for t in tasks]
        print(results)

asyncio.run(main())

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def demo_gather():
    """gather: 按顺序收集结果"""
    results = await asyncio.gather(
        fetch("A", 2),
        fetch("B", 1),
        fetch("C", 3),
    )
    print(f"gather 结果: {results}")
    # 输出: gather 结果: ['A 完成', 'B 完成', 'C 完成']
    # 注意：结果顺序与传入顺序一致，而非完成顺序

async def demo_wait():
    """wait: 获取完成和未完成的 Task 集合"""
    tasks = {
        asyncio.create_task(fetch("A", 2)),
        asyncio.create_task(fetch("B", 1)),
        asyncio.create_task(fetch("C", 3)),
    }
    done, pending = await asyncio.wait(tasks)
    print(f"完成: {[t.result() for t in done]}")
    # 输出顺序不确定，取决于完成顺序

asyncio.run(demo_gather())
asyncio.run(demo_wait())

import asyncio

async def demo_wait_strategies():
    async def fast():
        await asyncio.sleep(0.1)
        return "fast"

    async def medium():
        await asyncio.sleep(0.5)
        return "medium"

    async def slow():
        await asyncio.sleep(2.0)
        return "slow"

    async def failing():
        await asyncio.sleep(0.3)
        raise ValueError("出错了")

    # 策略1: FIRST_COMPLETED — 任一完成即返回
    tasks = {asyncio.create_task(fast()),
             asyncio.create_task(medium()),
             asyncio.create_task(slow())}
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
    print(f"FIRST_COMPLETED: 完成 {len(done)}, 待处理 {len(pending)}")
    # 输出: FIRST_COMPLETED: 完成 1, 待处理 2
    for t in pending:
        t.cancel()
    await asyncio.wait(pending)  # 等待取消完成

    # 策略2: FIRST_EXCEPTION — 第一个异常或全部完成
    tasks = {asyncio.create_task(fast()),
             asyncio.create_task(failing()),
             asyncio.create_task(slow())}
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_EXCEPTION
    )
    print(f"FIRST_EXCEPTION: 完成 {len(done)}, 待处理 {len(pending)}")
    for t in pending:
        t.cancel()
    await asyncio.wait(pending)

    # 策略3: ALL_COMPLETED — 全部完成（默认）
    tasks = {asyncio.create_task(fast()),
             asyncio.create_task(medium())}
    done, pending = await asyncio.wait(tasks)  # 默认 ALL_COMPLETED
    print(f"ALL_COMPLETED: 完成 {len(done)}, 待处理 {len(pending)}")
    # 输出: ALL_COMPLETED: 完成 2, 待处理 0

asyncio.run(demo_wait_strategies())

import asyncio

async def success():
    return "成功"

async def failure():
    raise ValueError("失败")

async def demo_exception_handling():
    # gather: return_exceptions=True 捕获异常
    results = await asyncio.gather(
        success(),
        failure(),
        success(),
        return_exceptions=True,
    )
    print(f"gather 结果: {results}")
    # 输出: ['成功', ValueError('失败'), '成功']

    # gather: return_exceptions=False（默认）抛出第一个异常
    try:
        await asyncio.gather(success(), failure(), success())
    except ValueError as e:
        print(f"gather 抛出异常: {e}")

    # wait: 异常 Task 归入 done 集合
    tasks = {
        asyncio.create_task(success()),
        asyncio.create_task(failure()),
    }
    done, pending = await asyncio.wait(tasks)
    for t in done:
        if t.exception():
            print(f"wait 发现异常: {t.exception()}")
        else:
            print(f"wait 正常结果: {t.result()}")

asyncio.run(demo_exception_handling())

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "完成"

async def demo_wait_for():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时！")

asyncio.run(demo_wait_for())
# 输出: 操作超时！

import asyncio

async def demo_timeout():
    async with asyncio.timeout(3.0):
        # 这里的所有操作共享同一个超时限制
        await asyncio.sleep(1)  # 剩余 2 秒
        await asyncio.sleep(1)  # 剩余 1 秒
        await asyncio.sleep(1)  # 超时！
    # 如果到达这里，说明所有操作在 3 秒内完成

async def demo_timeout_nested():
    """嵌套超时：内层超时先触发"""
    try:
        async with asyncio.timeout(5.0):   # 外层 5 秒
            async with asyncio.timeout(2.0):  # 内层 2 秒
                await asyncio.sleep(10)       # 内层先超时
    except asyncio.TimeoutError:
        print("内层超时触发")

asyncio.run(demo_timeout_nested())
# 输出: 内层超时触发

import asyncio
from typing import Any

async def fetch_with_retry(
    url: str,
    timeout: float = 5.0,
    retries: int = 3,
) -> Any:
    """带重试的超时请求"""
    last_error = None

    for attempt in range(retries):
        try:
            async with asyncio.timeout(timeout):
                # 模拟 HTTP 请求
                await asyncio.sleep(2)  # 模拟网络延迟
                return f"数据来自 {url}"
        except asyncio.TimeoutError:
            last_error = TimeoutError(f"第 {attempt + 1} 次请求超时")
            print(f"第 {attempt + 1} 次请求 {url} 超时，重试中...")

    raise last_error

async def main():
    try:
        result = await fetch_with_retry("http://slow-api.com", timeout=1.0)
    except TimeoutError as e:
        print(f"所有重试失败: {e}")

asyncio.run(main())

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io() -> str:
    """模拟阻塞 IO 操作"""
    time.sleep(2)  # 同步阻塞
    return "IO 完成"

def cpu_intensive(n: int) -> int:
    """模拟 CPU 密集型计算"""
    return sum(i * i for i in range(n))

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

    # 使用默认线程池执行阻塞 IO
    result = await loop.run_in_executor(None, blocking_io)
    print(f"阻塞 IO 结果: {result}")

    # 使用自定义线程池
    with ThreadPoolExecutor(max_workers=4) as pool:
        result = await loop.run_in_executor(pool, cpu_intensive, 10_000_000)
        print(f"CPU 计算结果: {result}")

asyncio.run(demo_executor())

import asyncio
import time

def blocking_call(url: str) -> str:
    """同步阻塞函数"""
    time.sleep(1)
    return f"来自 {url} 的数据"

async def demo_to_thread():
    # to_thread: 简洁的线程调用方式
    result = await asyncio.to_thread(blocking_call, "http://api.com")
    print(result)

    # 对比 run_in_executor
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_call, "http://api.com")
    print(result)

asyncio.run(demo_to_thread())

import asyncio
from concurrent.futures import ProcessPoolExecutor

def heavy_computation(n: int) -> int:
    """CPU 密集型计算，适合多进程"""
    total = 0
    for i in range(n):
        total += i * i
    return total

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

    # 使用进程池，绕过 GIL
    with ProcessPoolExecutor() as pool:
        # 并行执行多个计算任务
        results = await asyncio.gather(
            loop.run_in_executor(pool, heavy_computation, 5_000_000),
            loop.run_in_executor(pool, heavy_computation, 5_000_000),
            loop.run_in_executor(pool, heavy_computation, 5_000_000),
        )
        print(f"计算结果: {results}")

asyncio.run(demo_process_pool())

import asyncio
import threading

async def background_task(name: str):
    print(f"[{threading.current_thread().name}] {name} 开始")
    await asyncio.sleep(1)
    print(f"[{threading.current_thread().name}] {name} 完成")
    return f"{name} 结果"

def thread_worker(loop: asyncio.AbstractEventLoop):
    """从非事件循环线程安全地提交协程"""
    # asyncio.run_coroutine_threadsafe: 线程安全的协程提交
    future = asyncio.run_coroutine_threadsafe(
        background_task("来自线程的任务"), loop
    )
    # 可以等待结果（阻塞当前线程）
    result = future.result(timeout=5)
    print(f"线程获取结果: {result}")

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

    # 启动一个普通线程，从线程中提交协程到事件循环
    thread = threading.Thread(
        target=thread_worker, args=(loop,), daemon=True
    )
    thread.start()

    # 事件循环继续处理其他任务
    await asyncio.sleep(0.5)
    print("事件循环继续工作...")

    thread.join()

asyncio.run(main())

import asyncio
import logging
from typing import Any, Callable, Coroutine

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AsyncTaskManager:
    """异步任务管理器

    特性:
    - 并发控制（Semaphore）
    - 超时处理
    - 自动重试
    - 优雅关闭
    - 结果收集
    """

    def __init__(
        self,
        max_concurrency: int = 10,
        default_timeout: float = 30.0,
        default_retries: int = 3,
    ):
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._default_timeout = default_timeout
        self._default_retries = default_retries
        self._tasks: list[asyncio.Task] = []
        self._results: dict[str, Any] = {}
        self._errors: dict[str, Exception] = {}

    async def submit(
        self,
        name: str,
        coro_func: Callable[..., Coroutine],
        *args: Any,
        timeout: float | None = None,
        retries: int | None = None,
        **kwargs: Any,
    ) -> None:
        """提交一个任务到管理器"""
        task = asyncio.create_task(
            self._run_with_control(
                name, coro_func, *args,
                timeout=timeout or self._default_timeout,
                retries=retries if retries is not None else self._default_retries,
                **kwargs,
            )
        )
        self._tasks.append(task)

    async def _run_with_control(
        self,
        name: str,
        coro_func: Callable[..., Coroutine],
        *args: Any,
        timeout: float,
        retries: int,
        **kwargs: Any,
    ) -> Any:
        """带并发控制、超时和重试的任务执行"""
        async with self._semaphore:
            last_error: Exception | None = None

            for attempt in range(1, retries + 1):
                try:
                    async with asyncio.timeout(timeout):
                        result = await coro_func(*args, **kwargs)
                        self._results[name] = result
                        logger.info(f"任务 [{name}] 成功 (第 {attempt} 次)")
                        return result

                except asyncio.TimeoutError:
                    last_error = TimeoutError(
                        f"任务 [{name}] 第 {attempt} 次超时 ({timeout}s)"
                    )
                    logger.warning(str(last_error))

                except Exception as e:
                    last_error = e
                    logger.warning(f"任务 [{name}] 第 {attempt} 次失败: {e}")

            self._errors[name] = last_error
            raise last_error  # type: ignore

    async def wait_all(self) -> tuple[dict, dict]:
        """等待所有任务完成，返回 (结果, 错误)"""
        if not self._tasks:
            return self._results, self._errors

        results = await asyncio.gather(*self._tasks, return_exceptions=True)
        return self._results, self._errors

    async def graceful_shutdown(self, timeout: float = 10.0) -> None:
        """优雅关闭：取消所有未完成任务"""
        pending = [t for t in self._tasks if not t.done()]
        if not pending:
            return

        logger.info(f"正在取消 {len(pending)} 个未完成任务...")
        for task in pending:
            task.cancel()

        try:
            async with asyncio.timeout(timeout):
                await asyncio.gather(*pending, return_exceptions=True)
        except asyncio.TimeoutError:
            logger.warning("部分任务未能在超时内完成取消")

        logger.info("优雅关闭完成")


# 使用示例
async def fetch_api(url: str) -> str:
    """模拟 API 请求"""
    await asyncio.sleep(1)
    if "fail" in url:
        raise ConnectionError(f"连接 {url} 失败")
    return f"数据来自 {url}"


async def main():
    manager = AsyncTaskManager(max_concurrency=3, default_timeout=5.0)

    # 提交多个任务
    urls = [
        "http://api1.com/data",
        "http://api2.com/data",
        "http://fail-api.com/data",  # 这个会失败
        "http://api3.com/data",
        "http://api4.com/data",
    ]

    for i, url in enumerate(urls):
        await manager.submit(f"task-{i}", fetch_api, url)

    # 等待所有任务完成
    results, errors = await manager.wait_all()

    print(f"\n成功: {len(results)} 个")
    print(f"失败: {len(errors)} 个")
    for name, error in errors.items():
        print(f"  {name}: {error}")

asyncio.run(main())

import asyncio
import time
from typing import Any


class HybridService:
    """混合同步与异步的服务类

    场景：已有同步数据库客户端，需要集成到异步 Web 服务中
    """

    def __init__(self, db_url: str):
        self.db_url = db_url
        self._cache: dict[str, Any] = {}

    # ---- 同步方法（已有代码） ----

    def _sync_query(self, sql: str) -> list[dict]:
        """同步数据库查询（模拟已有代码）"""
        time.sleep(0.5)  # 模拟数据库 IO
        return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

    def _sync_cache_get(self, key: str) -> Any | None:
        """同步缓存读取"""
        return self._cache.get(key)

    def _sync_cache_set(self, key: str, value: Any) -> None:
        """同步缓存写入"""
        self._cache[key] = value

    # ---- 异步包装层 ----

    async def query(self, sql: str) -> list[dict]:
        """异步数据库查询（包装同步方法）"""
        # 先查缓存
        cache_key = f"query:{hash(sql)}"
        cached = await asyncio.to_thread(self._sync_cache_get, cache_key)
        if cached is not None:
            return cached

        # 缓存未命中，查询数据库
        result = await asyncio.to_thread(self._sync_query, sql)

        # 写入缓存
        await asyncio.to_thread(self._sync_cache_set, cache_key, result)
        return result

    async def batch_query(self, sqls: list[str]) -> list[list[dict]]:
        """批量异步查询"""
        tasks = [self.query(sql) for sql in sqls]
        return await asyncio.gather(*tasks)

    # ---- 纯异步方法 ----

    async def fetch_remote(self, url: str) -> str:
        """纯异步 HTTP 请求（使用异步库）"""
        await asyncio.sleep(0.3)  # 模拟 aiohttp 请求
        return f"远程数据来自 {url}"

    async def process(self, sql: str, api_url: str) -> dict:
        """混合处理：同时查询数据库和远程 API"""
        # 并行执行同步（线程化）和异步操作
        db_result, api_result = await asyncio.gather(
            self.query(sql),
            self.fetch_remote(api_url),
        )
        return {
            "database": db_result,
            "remote": api_result,
        }


async def main():
    service = HybridService("postgresql://localhost/mydb")

    # 单次查询
    result = await service.query("SELECT * FROM users")
    print(f"查询结果: {result}")

    # 批量查询
    results = await service.batch_query([
        "SELECT * FROM users",
        "SELECT * FROM orders",
    ])
    print(f"批量查询: {len(results)} 个结果集")

    # 混合处理
    combined = await service.process(
        "SELECT * FROM users",
        "http://api.example.com/data",
    )
    print(f"混合结果: {combined}")

asyncio.run(main())

import asyncio
import time

# 错误示范
async def bad_example():
    print("开始")
    time.sleep(3)       # 阻塞整个事件循环！
    await asyncio.sleep(1)
    print("结束")

# 正确做法
async def good_example():
    print("开始")
    await asyncio.sleep(3)  # 非阻塞等待
    await asyncio.sleep(1)
    print("结束")

# 如果必须调用同步阻塞函数
async def with_executor():
    print("开始")
    await asyncio.to_thread(time.sleep, 3)  # 在线程中执行
    await asyncio.sleep(1)
    print("结束")

import asyncio

async def important_work():
    await asyncio.sleep(1)
    return "重要结果"

async def bad():
    # Task 没有保存引用，可能被 GC 回收
    asyncio.create_task(important_work())
    await asyncio.sleep(2)
    # 可能看到 "Task was destroyed but it is pending!" 警告

async def good():
    task = asyncio.create_task(important_work())
    # 方式1: 等待完成
    result = await task
    # 方式2: 保存到集合中
    # background_tasks = set()
    # task = asyncio.create_task(important_work())
    # background_tasks.add(task)
    # task.add_done_callback(background_tasks.discard)