并发与异步编程¶

import threading
import time

def cpu_bound_task():
    """CPU 密集型任务"""
    count = 0
    for _ in range(50_000_000):
        count += 1
    return count

# 单线程
start = time.time()
cpu_bound_task()
cpu_bound_task()
print(f"单线程: {time.time() - start:.2f}s")

# 多线程（并不会更快！因为 GIL）
start = time.time()
t1 = threading.Thread(target=cpu_bound_task)  # threading.Thread创建新线程执行并发任务
t2 = threading.Thread(target=cpu_bound_task)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程: {time.time() - start:.2f}s")

# 输出示例：
# 单线程: 4.20s
# 多线程: 4.35s  ← 甚至可能更慢（线程切换开销）

import threading
import time

# 方式一：传入函数
def worker(name, delay):
    print(f"线程 {name} 开始")
    time.sleep(delay)
    print(f"线程 {name} 完成")

t1 = threading.Thread(target=worker, args=("A", 2))
t2 = threading.Thread(target=worker, args=("B", 1))

t1.start()          # 启动线程（非阻塞）
t2.start()

t1.join()           # 等待线程完成（阻塞）
t2.join()
print("所有线程完成")

# 方式二：继承 Thread 类
class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None

    def run(self):
        # 模拟下载
        print(f"正在下载: {self.url}")
        time.sleep(1)
        self.result = f"{self.url} 的内容"

t = DownloadThread("https://example.com")
t.start()
t.join()
print(t.result)

# 守护线程（主线程结束时自动终止）
t = threading.Thread(target=worker, args=("daemon", 10), daemon=True)
t.start()
# 主线程结束时，daemon 线程会被强制终止

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(1_000_000):
        with lock:              # 推荐使用 with 语句（自动释放）
            counter += 1

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()
print(f"counter = {counter}")   # 2000000（正确结果）
# 如果不加锁，结果可能小于 2000000

import threading

rlock = threading.RLock()

def recursive_function(n):
    with rlock:                 # 同一线程可以多次获取
        if n > 0:
            print(f"depth: {n}")
            recursive_function(n - 1)

# Lock 在同一线程重复获取会死锁，RLock 不会

import threading
import time

# 限制同时运行的线程数
semaphore = threading.Semaphore(3)  # 最多3个线程同时执行

def limited_task(name):
    with semaphore:
        print(f"任务 {name} 开始执行")
        time.sleep(2)
        print(f"任务 {name} 完成")

threads = [threading.Thread(target=limited_task, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# 任何时刻最多只有 3 个任务在执行

import threading
import time

event = threading.Event()

def waiter():
    print("等待信号...")
    event.wait()            # 阻塞，直到 event 被 set
    print("收到信号！开始工作")

def sender():
    time.sleep(2)
    print("发送信号")
    event.set()             # 唤醒所有等待的线程

threading.Thread(target=waiter).start()
threading.Thread(target=sender).start()

import threading
import time

condition = threading.Condition()
items = []

def producer():
    for i in range(5):
        time.sleep(0.5)
        with condition:
            items.append(f"item-{i}")
            print(f"生产: item-{i}")
            condition.notify()      # 通知消费者

def consumer():
    while True:
        with condition:
            while not items:
                condition.wait()    # 等待通知
            item = items.pop(0)
            print(f"消费: {item}")

threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()

import threading
import queue
import time

q = queue.Queue(maxsize=10)     # 线程安全的队列

def producer(q):
    for i in range(20):
        item = f"product-{i}"
        q.put(item)             # 如果队列满了会阻塞
        print(f"[生产者] 生产了: {item}")
        time.sleep(0.1)
    q.put(None)                 # 毒丸（Poison Pill）标记结束

def consumer(q):
    while True:
        item = q.get()          # 如果队列空了会阻塞
        if item is None:        # 收到结束信号
            break
        print(f"[消费者] 消费了: {item}")
        time.sleep(0.2)
        q.task_done()           # 标记任务完成

producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))

producer_thread.start()
consumer_thread.start()
producer_thread.join()
consumer_thread.join()
print("全部完成")

import threading
import queue
import time
import random

class ProducerConsumer:
    def __init__(self, buffer_size=5, num_producers=2, num_consumers=3):
        self.q = queue.Queue(maxsize=buffer_size)
        self.num_producers = num_producers
        self.num_consumers = num_consumers
        self.finished = threading.Event()

    def produce(self, producer_id):
        """生产者"""
        for i in range(5):
            item = f"P{producer_id}-Item{i}"
            self.q.put(item)
            print(f"[生产者-{producer_id}] 生产: {item} (队列大小: {self.q.qsize()})")
            time.sleep(random.uniform(0.1, 0.3))
        print(f"[生产者-{producer_id}] 完成生产")

    def consume(self, consumer_id):
        """消费者"""
        while not self.finished.is_set() or not self.q.empty():
            try:
                item = self.q.get(timeout=0.5)
                print(f"[消费者-{consumer_id}] 消费: {item}")
                time.sleep(random.uniform(0.1, 0.5))
                self.q.task_done()
            except queue.Empty:
                continue
        print(f"[消费者-{consumer_id}] 退出")

    def run(self):
        producers = [
            threading.Thread(target=self.produce, args=(i,))
            for i in range(self.num_producers)
        ]
        consumers = [
            threading.Thread(target=self.consume, args=(i,))
            for i in range(self.num_consumers)
        ]

        for t in producers + consumers:
            t.start()

        for t in producers:
            t.join()

        self.q.join()           # 等待所有任务被处理
        self.finished.set()     # 通知消费者退出

        for t in consumers:
            t.join()

        print("所有生产和消费完成")

if __name__ == "__main__":
    pc = ProducerConsumer()
    pc.run()

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def download(url):
    """模拟下载任务"""
    print(f"开始下载: {url}")
    time.sleep(1)
    return f"{url} 下载完成 ({len(url)} bytes)"

urls = [f"https://example.com/page/{i}" for i in range(10)]

# 使用线程池（推荐方式）
with ThreadPoolExecutor(max_workers=5) as executor:
    # 方式一：map（保持顺序）
    results = executor.map(download, urls)
    for result in results:
        print(result)

# 方式二：submit + as_completed（按完成顺序获取）
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(download, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(result)
        except Exception as e:
            print(f"{url} 出错: {e}")

import multiprocessing
import time
import os

def cpu_bound_task(n):
    """CPU 密集型任务"""
    print(f"进程 {os.getpid()} 开始计算")
    result = sum(i * i for i in range(n))
    print(f"进程 {os.getpid()} 完成")
    return result

if __name__ == "__main__":
    # 创建进程
    p1 = multiprocessing.Process(target=cpu_bound_task, args=(10_000_000,))
    p2 = multiprocessing.Process(target=cpu_bound_task, args=(10_000_000,))

    start = time.time()
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print(f"多进程耗时: {time.time() - start:.2f}s")

    # 对比单进程
    start = time.time()
    cpu_bound_task(10_000_000)
    cpu_bound_task(10_000_000)
    print(f"单进程耗时: {time.time() - start:.2f}s")

    # 多进程真的更快！因为每个进程有自己的 GIL

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(f"item-{i}")
    q.put(None)    # 结束信号

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费: {item}")

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

from multiprocessing import Process, Pipe

def sender(conn):
    conn.send("Hello from sender!")
    conn.send([1, 2, 3])
    conn.close()

def receiver(conn):
    while True:
        try:
            msg = conn.recv()
            print(f"收到: {msg}")
        except EOFError:
            break

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p1 = Process(target=sender, args=(child_conn,))
    p2 = Process(target=receiver, args=(parent_conn,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

from multiprocessing import Process, Value, Array

def modify(shared_val, shared_arr):
    # ⚠️ 并发安全提示：Value 的 += 操作不是原子的（先读后写），多进程并发时会产生竞态条件。
    # 应使用 shared_val.get_lock() 加锁：
    #   with shared_val.get_lock():
    #       shared_val.value += 1
    shared_val.value += 1
    for i in range(len(shared_arr)):
        shared_arr[i] *= 2

if __name__ == "__main__":
    # Value：共享单个值（'i' 表示整数，'d' 表示浮点数）
    val = Value('i', 0)
    # Array：共享数组
    arr = Array('i', [1, 2, 3, 4, 5])

    p = Process(target=modify, args=(val, arr))
    p.start()
    p.join()

    print(f"Value: {val.value}")     # 1
    print(f"Array: {list(arr)}")     # [2, 4, 6, 8, 10]

from multiprocessing import Pool
from concurrent.futures import ProcessPoolExecutor
import time

def heavy_computation(n):
    """模拟CPU密集计算"""
    return sum(i ** 2 for i in range(n))

if __name__ == "__main__":
    numbers = [5_000_000] * 8

    # 方式一：multiprocessing.Pool
    start = time.time()
    with Pool(processes=4) as pool:
        results = pool.map(heavy_computation, numbers)
    print(f"Pool: {time.time() - start:.2f}s")

    # 方式二：ProcessPoolExecutor（推荐，与 ThreadPoolExecutor 接口一致）
    start = time.time()
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(heavy_computation, numbers))
    print(f"ProcessPoolExecutor: {time.time() - start:.2f}s")

    # 对比单进程
    start = time.time()
    results = [heavy_computation(n) for n in numbers]
    print(f"单进程: {time.time() - start:.2f}s")

# 阶段一：生成器（Python 2.5+）
def simple_generator():
    yield 1
    yield 2
    yield 3

# 阶段二：yield from（Python 3.3+）
def sub_gen():
    yield "a"
    yield "b"

def main_gen():
    yield from sub_gen()    # 委托给子生成器
    yield "c"

# 阶段三：@asyncio.coroutine + yield from（Python 3.4）
import asyncio

@asyncio.coroutine
def old_style_coroutine():
    yield from asyncio.sleep(1)
    return "done"

# 阶段四：async/await（Python 3.5+）✨ 现代方式
async def modern_coroutine():
    await asyncio.sleep(1)
    return "done"

import asyncio

async def say_hello(name, delay):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")

# Python 3.7+ 推荐方式
async def main():
    # 这三个协程会并发执行
    await asyncio.gather(
        say_hello("Alice", 2),
        say_hello("Bob", 1),
        say_hello("Charlie", 3),
    )

asyncio.run(main())  # asyncio.run()创建事件循环并运行顶层协程，是异步程序的入口
# 输出（按完成顺序）：
# Hello, Bob!        (1秒后)
# Hello, Alice!      (2秒后)
# Hello, Charlie!    (3秒后)
# 总耗时约3秒，而非6秒！

import asyncio

# async def 定义协程函数
async def fetch_data(url):
    print(f"开始获取 {url}")
    await asyncio.sleep(1)          # 模拟网络请求（非阻塞）
    return f"{url} 的数据"

# await 只能在 async 函数中使用
async def process():
    # await 暂停当前协程，等待结果
    data = await fetch_data("https://api.example.com")
    print(f"处理: {data}")

asyncio.run(process())

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成(耗时{delay}s)"

async def main():
    # 1. create_task：创建任务（立即调度执行）
    t1 = asyncio.create_task(task("A", 2))
    t2 = asyncio.create_task(task("B", 1))
    result1 = await t1
    result2 = await t2
    print(result1, result2)

    # 2. gather：并发运行多个协程，收集所有结果
    results = await asyncio.gather(
        task("C", 2),
        task("D", 1),
        task("E", 3),
    )
    print(results)  # ['C 完成(耗时2s)', 'D 完成(耗时1s)', 'E 完成(耗时3s)']

    # 3. wait：并发运行，更灵活的控制
    tasks = [asyncio.create_task(task(f"T{i}", i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(tasks, timeout=2.5)
    for t in done:
        print(f"完成: {t.result()}")
    for t in pending:
        print(f"未完成: {t.get_name()}")
        t.cancel()

    # 4. sleep：非阻塞等待
    await asyncio.sleep(1)

    # 5. wait_for：设置超时
    try:
        result = await asyncio.wait_for(task("slow", 10), timeout=3)
    except asyncio.TimeoutError:
        print("任务超时！")

    # 6. as_completed：按完成顺序获取结果
    coros = [task(f"X{i}", 3 - i) for i in range(3)]
    for coro in asyncio.as_completed(coros):
        result = await coro
        print(f"按完成顺序: {result}")

asyncio.run(main())

import asyncio

class AsyncResource:
    async def __aenter__(self):  # 异步上下文管理器入口，对应 async with 语句
        print("获取资源")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):  # 异步退出：exc_type=异常类型, exc_val=异常值, exc_tb=回溯信息
        print("释放资源")
        await asyncio.sleep(0.1)

    async def do_something(self):
        print("使用资源")

async def main():
    async with AsyncResource() as resource:  # async with异步上下文管理器，自动管理异步资源的获取和释放
        await resource.do_something()

asyncio.run(main())
# 获取资源
# 使用资源
# 释放资源

import asyncio

class AsyncCounter:
    def __init__(self, stop):
        self.current = 0
        self.stop = stop

    def __aiter__(self):  # 返回异步迭代器自身，使对象可用于 async for
        return self

    async def __anext__(self):  # 异步迭代器的"取下一个"方法
        if self.current >= self.stop:
            raise StopAsyncIteration  # 抛出此异常表示异步迭代结束（类似同步的StopIteration）
        self.current += 1
        await asyncio.sleep(0.1)
        return self.current

async def main():
    async for num in AsyncCounter(5):
        print(num)

asyncio.run(main())

# 异步生成器（Python 3.6+）
async def async_range(start, stop):
    for i in range(start, stop):
        await asyncio.sleep(0.1)
        yield i

async def main2():
    async for i in async_range(1, 6):
        print(i)

asyncio.run(main2())

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """异步获取单个URL"""
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
    ]

    start = time.time()

    # 使用 aiohttp 的 ClientSession
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    elapsed = time.time() - start
    print(f"异步请求 {len(urls)} 个URL，耗时: {elapsed:.2f}s")
    # 5个延迟1秒的请求，同步需要5秒，异步只需约1秒

    # 限制并发数量
    semaphore = asyncio.Semaphore(3)

    async def limited_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [limited_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

# pip install aiohttp
asyncio.run(main())

import asyncio
import aiofiles

async def async_file_operations():
    # 异步写文件
    async with aiofiles.open("output.txt", mode="w") as f:
        await f.write("Hello, async world!\n")
        await f.write("Line 2\n")

    # 异步读文件
    async with aiofiles.open("output.txt", mode="r") as f:
        content = await f.read()
        print(content)

    # 异步逐行读取
    async with aiofiles.open("output.txt", mode="r") as f:
        async for line in f:
            print(line.strip())

# pip install aiofiles
asyncio.run(async_file_operations())

import asyncio

# 示例：使用 asyncpg（PostgreSQL 异步驱动）
# pip install asyncpg

async def db_operations():
    import asyncpg

    # 创建连接池
    pool = await asyncpg.create_pool(
        host="localhost",
        database="mydb",
        user="user",
        password="password",
        min_size=5,
        max_size=20,
    )

    async with pool.acquire() as conn:
        # 查询
        rows = await conn.fetch("SELECT * FROM users WHERE active = $1", True)
        for row in rows:
            print(dict(row))

        # 插入
        await conn.execute(
            "INSERT INTO users(name, email) VALUES($1, $2)",
            "Alice", "alice@example.com"
        )

        # 事务
        async with conn.transaction():
            await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = $1", 1)
            await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = $1", 2)

    await pool.close()

# 常见异步数据库驱动：
# PostgreSQL: asyncpg, aiopg
# MySQL: aiomysql
# SQLite: aiosqlite
# ORM: SQLAlchemy 2.0 (支持 async), Tortoise ORM

import time
import threading
import multiprocessing
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import requests  # pip install requests

# ============================================
# IO 密集型任务对比
# ============================================

URL = "https://httpbin.org/delay/1"  # 每个请求延迟1秒
NUM_REQUESTS = 10

# 1. 同步方式
def sync_fetch():
    results = []
    start = time.time()
    for _ in range(NUM_REQUESTS):
        resp = requests.get(URL)
        results.append(resp.status_code)
    elapsed = time.time() - start
    print(f"同步: {elapsed:.2f}s")

# 2. 多线程
def threaded_fetch():
    start = time.time()
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(lambda _: requests.get(URL).status_code, range(NUM_REQUESTS)))
    elapsed = time.time() - start
    print(f"多线程: {elapsed:.2f}s")

# 3. 异步
async def async_fetch():
    import aiohttp
    start = time.time()
    async with aiohttp.ClientSession() as session:
        async def _fetch(url):
            async with session.get(url) as resp:  # async with 确保响应被正确释放
                return resp.status
        tasks = [_fetch(URL) for _ in range(NUM_REQUESTS)]
        results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    print(f"异步: {elapsed:.2f}s")

# 预期结果（10个延迟1秒的请求）：
# 同步:   ~10.0s
# 多线程: ~2.0s  (5个worker)
# 异步:   ~1.0s  (全部并发)

# ============================================
# CPU 密集型任务对比
# ============================================

def cpu_task(n):
    """计算密集"""
    return sum(i ** 2 for i in range(n))

def compare_cpu():
    N = 5_000_000
    COUNT = 4

    # 单进程
    start = time.time()
    for _ in range(COUNT):
        cpu_task(N)
    print(f"单进程 CPU 密集: {time.time() - start:.2f}s")

    # 多线程（不会更快）
    start = time.time()
    with ThreadPoolExecutor(max_workers=4) as executor:
        list(executor.map(cpu_task, [N] * COUNT))
    print(f"多线程 CPU 密集: {time.time() - start:.2f}s")

    # 多进程（真正加速）
    start = time.time()
    with ProcessPoolExecutor(max_workers=4) as executor:
        list(executor.map(cpu_task, [N] * COUNT))
    print(f"多进程 CPU 密集: {time.time() - start:.2f}s")

"""
三种爬虫实现方式对比：同步 vs 多线程 vs 异步
"""

import time
import requests
import threading
from concurrent.futures import ThreadPoolExecutor
import asyncio

# ==========================================
# 版本一：同步爬虫
# ==========================================
def sync_crawler(urls):
    """同步爬虫"""
    results = []
    start = time.time()
    for url in urls:
        try:
            resp = requests.get(url, timeout=10)
            results.append({
                "url": url,
                "status": resp.status_code,
                "length": len(resp.text),
            })
        except Exception as e:
            results.append({"url": url, "error": str(e)})
    elapsed = time.time() - start
    print(f"同步爬虫: {elapsed:.2f}s, 成功: {sum(1 for r in results if 'status' in r)}/{len(urls)}")
    return results

# ==========================================
# 版本二：多线程爬虫
# ==========================================
def threaded_crawler(urls, max_workers=10):
    """多线程爬虫"""
    results = []
    lock = threading.Lock()

    def fetch(url):
        try:
            resp = requests.get(url, timeout=10)
            result = {"url": url, "status": resp.status_code, "length": len(resp.text)}
        except Exception as e:
            result = {"url": url, "error": str(e)}
        with lock:
            results.append(result)

    start = time.time()
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(fetch, urls)
    elapsed = time.time() - start
    print(f"多线程爬虫: {elapsed:.2f}s, 成功: {sum(1 for r in results if 'status' in r)}/{len(urls)}")
    return results

# ==========================================
# 版本三：异步爬虫
# ==========================================
async def async_crawler(urls, max_concurrent=20):
    """异步爬虫"""
    import aiohttp

    results = []
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch(session, url):
        async with semaphore:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    text = await resp.text()
                    return {"url": url, "status": resp.status, "length": len(text)}
            except Exception as e:
                return {"url": url, "error": str(e)}

    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    print(f"异步爬虫: {elapsed:.2f}s, 成功: {sum(1 for r in results if 'status' in r)}/{len(urls)}")
    return results

# ==========================================
# 对比运行
# ==========================================
if __name__ == "__main__":
    # 准备测试 URL 列表
    test_urls = [f"https://httpbin.org/get?id={i}" for i in range(20)]

    print("=" * 50)
    print(f"爬取 {len(test_urls)} 个 URL")
    print("=" * 50)

    # 同步
    sync_crawler(test_urls)

    # 多线程
    threaded_crawler(test_urls, max_workers=10)

    # 异步（需要 pip install aiohttp）
    asyncio.run(async_crawler(test_urls, max_concurrent=20))

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def task(n):
    """通用任务"""
    time.sleep(0.1)
    return n ** 2

def run_with_executor(executor_class, max_workers, data):
    """通用执行器"""
    with executor_class(max_workers=max_workers) as executor:
        # submit：提交单个任务
        future = executor.submit(task, 42)
        print(f"单个结果: {future.result()}")

        # map：批量提交（保持顺序）
        results = list(executor.map(task, data))
        print(f"批量结果: {results[:5]}...")

if __name__ == "__main__":
    data = list(range(20))

    print("--- ThreadPoolExecutor ---")
    run_with_executor(ThreadPoolExecutor, 4, data)

    print("--- ProcessPoolExecutor ---")
    run_with_executor(ProcessPoolExecutor, 4, data)

    # 只需要改一个类名！接口完全一致

# ❌ 不能在同步函数中直接调用异步函数
def sync_function():
    result = await async_function()  # SyntaxError!

# ✅ 正确方式
async def correct_way():
    result = await async_function()

# 或者在同步上下文中运行
import asyncio
result = asyncio.run(async_function())

# ❌ 忘记 await，协程不会执行
async def main():
    asyncio.sleep(1)        # 没有 await！这只是创建了协程对象
    print("立刻打印了")

# ✅ 正确
async def main():
    await asyncio.sleep(1)
    print("等待1秒后打印")

# ❌ time.sleep 会阻塞整个事件循环
async def bad_example():
    time.sleep(5)           # 阻塞！其他协程也无法执行

# ✅ 使用异步版本
async def good_example():
    await asyncio.sleep(5)  # 非阻塞

# ✅ 如果必须使用阻塞操作，放到线程池中
async def necessary_blocking():
    loop = asyncio.get_running_loop()  # Python 3.10+推荐，替代已弃用的get_event_loop()
    result = await loop.run_in_executor(None, requests.get, "https://example.com")

# ❌ 非线程安全
counter = 0
def increment():
    global counter
    for _ in range(1_000_000):
        counter += 1       # 读-改-写不是原子操作！

# ✅ 使用锁
lock = threading.Lock()
def safe_increment():
    global counter
    for _ in range(1_000_000):
        with lock:
            counter += 1

sem = asyncio.Semaphore(10)
async def limited_task():
    async with sem:
        await do_something()

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(coro1())
        task2 = tg.create_task(coro2())
    # 所有任务完成后才会离开 with 块

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    # 在默认线程池中运行阻塞函数
    result = await loop.run_in_executor(None, blocking_function, arg1, arg2)

    # 或指定线程池
    from concurrent.futures import ThreadPoolExecutor
    executor = ThreadPoolExecutor(max_workers=5)
    result = await loop.run_in_executor(executor, blocking_function, arg1, arg2)