import threading
import time
import requests

# 模拟 I/O 阻塞
def fetch_url(url):
    print(f"[线程 {threading.current_thread().name}] 抓取 {url}")
    resp = requests.get(url, timeout=5)
    print(f"[线程 {threading.current_thread().name}] 完成，状态码 {resp.status_code}")

urls = ["https://www.baidu.com", "https://www.zhihu.com", "https://www.bilibili.com"]
threads = []

for url in urls:
    t = threading.Thread(target=fetch_url, args=(url,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()  # 等待所有线程完成

print("所有任务完成！")

import multiprocessing as mp
import os

def cpu_bound_task(n):
    pid = os.getpid()
    print(f"进程 {pid} 开始计算 {n}...")
    total = sum(i * i for i in range(10**7))  # 模拟 CPU 密集
    print(f"进程 {pid} 完成")
    return total

if __name__ == "__main__":
    # 强制使用 spawn 模式（跨平台且安全，适合生产环境）
    mp.set_start_method('spawn', force=True)

    with mp.Pool(processes=4) as pool:
        results = pool.map(cpu_bound_task, range(4))

    print(f"所有结果: {results}")

import multiprocessing as mp

def producer(queue):
    for i in range(5):
        queue.put(f"数据 {i}")
    queue.put(None)  # 结束标志

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

if __name__ == "__main__":
    q = mp.Queue()  # 进程安全队列
    p1 = mp.Process(target=producer, args=(q,))
    p2 = mp.Process(target=consumer, args=(q,))
    p1.start(); p2.start()
    p1.join(); p2.join()

import asyncio
import aiohttp

async def fetch_url(session, url):
    print(f"开始抓取: {url}")
    async with session.get(url) as response:
        html = await response.text()
        print(f"完成抓取: {url}, 长度 {len(html)}")
        return len(html)

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ["https://www.baidu.com", "https://www.zhihu.com", "https://www.bilibili.com"]
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"全部结果: {results}")

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

import threading

count = 0
def no_lock_increment():
    global count
    for _ in range(100000):
        count += 1  # 非原子操作！

threads = [threading.Thread(target=no_lock_increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"预期 1,000,000，实际 {count}")  # 实际总是小于预期

lock = threading.Lock()
count = 0

def safe_increment():
    global count
    for _ in range(100000):
        with lock:  # 上下文管理器，自动 acquire/release
            count += 1

rlock = threading.RLock()

def recursive_func(n):
    with rlock:
        if n > 0:
            recursive_func(n - 1)  # 同一线程可重复进入

sem = threading.Semaphore(5)  # 最多 5 个并发

def call_third_party_api():
    with sem:  # 占用一个令牌
        # 调用外部接口，对方限制 QPS
        pass

event = threading.Event()

def worker():
    event.wait()  # 阻塞等待
    print("出发！")

for _ in range(10):
    threading.Thread(target=worker).start()

time.sleep(2)
event.set()  # 所有线程同时触发

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def background_job(data: str):
    await asyncio.sleep(10)  # 模拟耗时
    print(f"后台完成: {data}")

@app.post("/submit")
async def submit_task(data: str):
    # 创建后台任务，不等待
    asyncio.create_task(background_job(data))
    return {"status": "accepted", "msg": "任务已接收"}

from celery import Celery

app_celery = Celery('tasks', broker='redis://localhost:6379/0')

@app_celery.task
def celery_background_job(user_id):
    # 此任务在独立的 Worker 进程中执行，Web 重启不影响
    time.sleep(10)
    print(f"Celery 完成用户 {user_id}")

# 视图函数中调用
celery_background_job.delay(user_id)  # 异步调用，立即返回

# ❌ 严重错误：在协程中调用同步 requests.get
async def bad_coro():
    requests.get("https://xxx.com")  # 事件循环卡死 5 秒

# ✅ 正确做法：交给线程池隔离
async def good_coro():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, requests.get, "https://xxx.com")