并发、异步与多线程

  # 简单的 Flask 应用
from flask import Flask

app = Flask(__name__)

@app.route('/order')
def create_order():
    # 查询库存
    stock = db.query("SELECT stock FROM products WHERE id=1")
    if stock > 0:
        # 扣减库存
        db.execute("UPDATE products SET stock = stock - 1 WHERE id=1")
        # 创建订单
        db.execute("INSERT INTO orders ...")
        return "Order created!"
    return "Out of stock!"

# 启动：flask run
  

  # 使用 Gunicorn 多进程部署
gunicorn -w 4 -k sync app:app

# 4个 worker 进程，每个进程独立处理请求
  

  import redis

# 使用 Redis 分布式锁
lock = redis_client.lock("stock_lock", timeout=10)
if lock.acquire():
    try:
        stock = db.query("SELECT stock FROM products WHERE id=1")
        if stock > 0:
            db.execute("UPDATE products SET stock = stock - 1 WHERE id=1")
    finally:
        lock.release()
  

  # 使用 FastAPI + asyncio
from fastapi import FastAPI
import asyncio

app = FastAPI()

async def check_stock(product_id: int) -> int:
    # 异步查询数据库，不阻塞
    result = await db.fetch_one(
        "SELECT stock FROM products WHERE id = :id",
        {"id": product_id}
    )
    return result["stock"]

@app.get("/order")
async def create_order(product_id: int):
    # 并发检查库存和用户信息
    stock_task = check_stock(product_id)
    user_task = get_user_info(request.user_id)

    stock, user = await asyncio.gather(stock_task, user_task)

    if stock > 0:
        # 异步扣减库存
        await db.execute(
            "UPDATE products SET stock = stock - 1 WHERE id = :id",
            {"id": product_id}
        )
        return {"status": "success"}

    return {"status": "out_of_stock"}

# 启动：uvicorn main:app --workers 4
# 每个 worker 内可以处理数千个并发协程
  

  进程 A 的虚拟内存          进程 B 的虚拟内存
+----------------+        +----------------+
|  内核空间      |        |  内核空间      |  <-- 共享（只读）
|  (共享)        |        |  (共享)        |
+----------------+        +----------------+
|  栈空间        |        |  栈空间        |  <-- 独立
|  (向下增长)    |        |  (向下增长)    |
+----------------+        +----------------+
|  堆空间        |        |  堆空间        |  <-- 独立
|  (向上增长)    |        |  (向上增长)    |
+----------------+        +----------------+
|  数据段        |        |  数据段        |  <-- 独立
|  (.bss/.data)  |        |  (.bss/.data)  |
+----------------+        +----------------+
|  代码段        |        |  代码段        |  <-- 独立
|  (.text)       |        |  (.text)       |
+----------------+        +----------------+
  

  就绪队列                    运行中                    等待队列
+--------+                +--------+               +--------+
| 线程 B |  <-- 时间片到   | 线程 A |  <-- I/O请求  | 线程 C |
| 线程 D |                | (运行) |               | 线程 E |
| 线程 F |                +--------+               | (阻塞) |
+--------+                                         +--------+
    |                                                  |
    v                                                  v
调度器根据优先级选择下一个运行            I/O完成时移回就绪队列
  

  传统多线程                vs              协程模型

+------------+                       +------------+
|  线程 1    |                       |  事件循环   |
| (1MB栈)   |                       |  (调度器)   |
+------------+                       +------------+
     |                                     |
     v                                     v
+------------+                       +------------+
|  线程 2    |                       |  协程 A    |
| (1MB栈)   |                       | (几KB栈)   |
+------------+                       +------------+
     |                                     |
     v                                     v
+------------+                       +------------+
|  线程 3    |                       |  协程 B    |
| (1MB栈)   |                       | (几KB栈)   |
+------------+                       +------------+

开销：N MB                           开销：N KB
创建：~10μs                         创建：~100ns
切换：~1μs                          切换：~100ns
  

  import asyncio

async def fetch_data(url):
    # 遇到 await，协程挂起，让出 CPU
    response = await aiohttp.get(url)
    # I/O 完成后，事件循环唤醒协程，从这里继续执行
    return response.json()

async def main():
    # 创建 3 个协程任务
    tasks = [
        fetch_data("https://api1.example.com"),
        fetch_data("https://api2.example.com"),
        fetch_data("https://api3.example.com")
    ]
    # 并发执行，总耗时 ≈ 最慢的那个请求
    results = await asyncio.gather(*tasks)
    return results

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

  时间线 -------------------------------------------------------------------->

协程 A: [准备请求]--[await 挂起]=======[收到响应]--[处理数据]
                     |
协程 B:              [准备请求]--[await 挂起]=======[收到响应]--[处理数据]
                                  |
协程 C:                           [准备请求]--[await 挂起]=======[收到响应]
                                               |
                                               ↓
                                         所有 I/O 完成

说明：[ ] 表示 CPU 执行, === 表示 I/O 等待, | 表示协程切换
  

  import selectors
import heapq

class EventLoop:
    def __init__(self):
        self.selector = selectors.DefaultSelector()
        self.ready = []  # 就绪队列
        self.scheduled = []  # 定时任务队列
        self.current = None

    def run(self):
        while True:
            # 1. 处理定时任务
            now = time.time()
            while self.scheduled and self.scheduled[0][0] <= now:
                _, callback = heapq.heappop(self.scheduled)
                self.ready.append(callback)

            # 2. 等待 I/O 事件
            timeout = 0 if self.ready else 0.1
            events = self.selector.select(timeout)

            for key, mask in events:
                callback = key.data
                self.ready.append(callback)

            # 3. 执行就绪的回调
            while self.ready:
                callback = self.ready.popleft()
                callback()
  

  单核 CPU - 并发（Concurrent）
时间 →  1    2    3    4    5    6    7    8
任务 A: [执行][执行]      [执行][执行]
任务 B:      [执行][执行]      [执行][执行]

两个任务交替执行，宏观上"同时"推进

========================================

多核 CPU - 并行（Parallel）
时间 →  1    2    3    4    5    6    7    8
核心 1: [任务A][任务A][任务A][任务A]
核心 2: [任务B][任务B][任务B][任务B]

两个任务真正"同时"执行

========================================

现实中往往是：并发 + 并行
时间 →  1    2    3    4    5    6    7    8
核心 1: [A1][A1][B1][B1][C1][C1][D1][D1]
核心 2: [A2][A2][B2][B2][C2][C2][D2][D2]

多个任务先并发调度到不同核心，再在核心上并行执行
  

  package main

import (
    "fmt"
    "time"
)

// 生产者
func producer(ch chan<- int, id int) {
    for i := 0; i < 5; i++ {
        fmt.Printf("Producer %d sending: %d\n", id, i)
        ch <- i  // 发送数据到 channel
        time.Sleep(100 * time.Millisecond)
    }
}

// 消费者
func consumer(ch <-chan int, id int) {
    for val := range ch {  // 从 channel 接收数据
        fmt.Printf("Consumer %d received: %d\n", id, val)
    }
}

func main() {
    // 创建带缓冲的 channel
    ch := make(chan int, 10)

    // 启动 2 个生产者 goroutine
    for i := 0; i < 2; i++ {
        go producer(ch, i)
    }

    // 启动 2 个消费者 goroutine
    for i := 0; i < 2; i++ {
        go consumer(ch, i)
    }

    // 等待一段时间
    time.Sleep(3 * time.Second)
    close(ch)
}
  

  全局队列
+----------------+
|  G1  |  G2  |  G3  |
+----------------+

P0 的本地队列       P1 的本地队列       P2 的本地队列       P3 的本地队列
+----------+       +----------+       +----------+       +----------+
| G4 | G5  |       | G6 | G7  |       | G8 | G9  |       | G10| G11 |
+----------+       +----------+       +----------+       +----------+
    |                     |                     |                     |
    v                     v                     v                     v
+----------+       +----------+       +----------+       +----------+
|    M0    |       |    M1    |       |    M2    |       |    M3    |
| (OS线程) |       | (OS线程) |       | (OS线程) |       | (OS线程) |
+----------+       +----------+       +----------+       +----------+

调度策略：
1. 每个 P 维护一个本地 G 队列，减少锁竞争
2. P 从本地队列取 G 交给 M 执行
3. 本地队列空时，从其他 P"偷"一半的 G（Work Stealing）
4. 全局队列作为兜底，每隔一段时间检查一次
  

  import asyncio
import aiohttp
from typing import List, Dict
import time

class AsyncHTTPClient:
    """基于 asyncio 的高性能 HTTP 客户端"""

    def __init__(self, max_connections: int = 100, timeout: int = 30):
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        # 限制并发连接数，防止把对方服务打挂
        connector = aiohttp.TCPConnector(
            limit=max_connections,
            limit_per_host=10,  # 对单个域名的连接限制
            enable_cleanup_closed=True,
            force_close=True,
        )
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=self.timeout,
        )

    async def fetch(self, url: str, method: str = 'GET', **kwargs) -> Dict:
        """发送单个请求"""
        try:
            async with self.session.request(method, url, **kwargs) as response:
                return {
                    'url': url,
                    'status': response.status,
                    'data': await response.text(),
                    'error': None
                }
        except asyncio.TimeoutError:
            return {'url': url, 'status': None, 'data': None, 'error': 'Timeout'}
        except Exception as e:
            return {'url': url, 'status': None, 'data': None, 'error': str(e)}

    async def fetch_many(self, urls: List[str], concurrency: int = 10) -> List[Dict]:
        """并发获取多个 URL，限制并发数"""
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_with_limit(url):
            async with semaphore:
                return await self.fetch(url)

        # 并发执行所有请求
        tasks = [fetch_with_limit(url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

    async def close(self):
        await self.session.close()


# 使用示例
async def main():
    client = AsyncHTTPClient(max_connections=50)

    # 要抓取的 URL 列表
    urls = [
        "https://api.github.com/users/github",
        "https://api.github.com/users/google",
        "https://api.github.com/users/microsoft",
        # ... 更多 URL
    ] * 10  # 模拟 300 个请求

    start = time.time()
    results = await client.fetch_many(urls, concurrency=20)
    elapsed = time.time() - start

    # 统计结果
    success = sum(1 for r in results if r.get('status') == 200)
    failed = len(results) - success

    print(f"总请求数: {len(results)}")
    print(f"成功: {success}, 失败: {failed}")
    print(f"耗时: {elapsed:.2f}s")
    print(f"QPS: {len(results)/elapsed:.1f}")

    await client.close()

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

  package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"runtime"
	"time"

	"golang.org/x/sync/errgroup"
)

// Request/Response 结构
type OrderRequest struct {
	UserID    int64   `json:"user_id"`
	ProductID int64   `json:"product_id"`
	Quantity  int     `json:"quantity"`
	Price     float64 `json:"price"`
}

type OrderResponse struct {
	OrderID   int64   `json:"order_id"`
	Status    string  `json:"status"`
	Total     float64 `json:"total"`
	CreatedAt string  `json:"created_at"`
}

// 模拟数据库操作
type Database struct {
	orders map[int64]*OrderResponse
	mutex  chan struct{}
}

func NewDatabase() *Database {
	db := &Database{
		orders: make(map[int64]*OrderResponse),
		mutex:  make(chan struct{}, 1), // 模拟互斥锁
	}
	return db
}

func (db *Database) CreateOrder(ctx context.Context, req *OrderRequest) (*OrderResponse, error) {
	// 获取锁
	select {
	case db.mutex <- struct{}{}:
		defer func() { <-db.mutex }()
	case <-ctx.Done():
		return nil, ctx.Err()
	}

	// 模拟数据库操作延迟
	select {
	case <-time.After(50 * time.Millisecond):
	case <-ctx.Done():
		return nil, ctx.Err()
	}

	order := &OrderResponse{
		OrderID:   time.Now().UnixNano(),
		Status:    "created",
		Total:     req.Price * float64(req.Quantity),
		CreatedAt: time.Now().Format(time.RFC3339),
	}
	db.orders[order.OrderID] = order
	return order, nil
}

// HTTP 处理器
type Handler struct {
	db *Database
}

func NewHandler(db *Database) *Handler {
	return &Handler{db: db}
}

func (h *Handler) CreateOrder(w http.ResponseWriter, r *http.Request) {
	// 设置请求超时
	ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
	defer cancel()

	var req OrderRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	order, err := h.db.CreateOrder(ctx, &req)
	if err != nil {
		if err == context.DeadlineExceeded {
			http.Error(w, "Request timeout", http.StatusGatewayTimeout)
			return
		}
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(order)
}

func (h *Handler) Health(w http.ResponseWriter, r *http.Request) {
	info := map[string]interface{}{
		"status":    "ok",
		"goroutine": runtime.NumGoroutine(),
		"cpu":       runtime.NumCPU(),
		"version":   runtime.Version(),
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(info)
}

// 批量处理示例
func BatchProcess(ctx context.Context, items []int) ([]int, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(10) // 限制并发数为 10

	results := make([]int, len(items))

	for i, item := range items {
		i, item := i, item // 避免闭包陷阱
		g.Go(func() error {
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
				// 模拟处理
				time.Sleep(100 * time.Millisecond)
				results[i] = item * 2
				return nil
			}
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

func main() {
	// 初始化数据库
	db := NewDatabase()

	// 创建处理器
	handler := NewHandler(db)

	// 设置路由
	mux := http.NewServeMux()
	mux.HandleFunc("/order", handler.CreateOrder)
	mux.HandleFunc("/health", handler.Health)

	// 创建服务器
	server := &http.Server{
		Addr:         ":8080",
		Handler:      mux,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 10 * time.Second,
		IdleTimeout:  120 * time.Second,
	}

	fmt.Println("Server starting on :8080")
	fmt.Printf("Go version: %s\n", runtime.Version())
	fmt.Printf("CPU cores: %d\n", runtime.NumCPU())

	if err := server.ListenAndServe(); err != nil {
		log.Fatal(err)
	}
}
  

  阶段 1: 基础理解
    ├── 理解进程/线程的基本概念
    ├── 学习同步原语（锁、信号量、条件变量）
    └── 编写简单的多线程程序

阶段 2: 深入原理
    ├── 理解内存模型和可见性
    ├── 学习无锁编程和原子操作
    ├── 理解线程池和工作窃取
    └── 分析死锁和竞态条件

阶段 3: 高级应用
    ├── 掌握协程和异步编程
    ├── 学习 Go/Python/Rust 的并发模型
    ├── 理解分布式系统中的并发
    └── 性能调优和容量规划

阶段 4: 专家水平
    ├── 设计高并发系统架构
    ├── 解决复杂的并发 Bug
    ├── 开发并发编程框架
    └── 分享和传播并发知识