Python asyncio 异步编程实战：从协程到任务调度 前言Python 的 asyncio 是编写并发代码的标准库，使用 async/await 语法实现单线程下的高并发 IO 操作。本文将从协程基础讲到实际应用，帮你掌握 Python 异步编程。一、同步 vs 异步1.1 同步代码的问题import requests import time def fetch_url(url): resp = requests.get(url) return resp.status_code start = time.time() for url in ["https://httpbin.org/delay/1"] * 10: fetch_url(url) print(f"同步耗时: {time.time() - start:.2f}s") # ~10s10 个请求串行执行，每个等待 1 秒，总共 10 秒。1.2 异步的优势import asyncio import aiohttp import time async def fetch_url(session, url): async with session.get(url) as resp: return resp.status async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, "https://httpbin.org/delay/1") for _ in range(10)] results = await asyncio.gather(*tasks) print(results) start = time.time() asyncio.run(main()) print(f"异步耗时: {time.time() - start:.2f}s") # ~1s10 个请求并发执行，总共约 1 秒。二、协程基础2.1 定义和调用协程import asyncio async def hello(): print("Hello") await asyncio.sleep(1) print("World") asyncio.run(hello())2.2 await 的含义await 只能在 async 函数内使用，它会：暂停当前协程的执行将控制权交还给事件循环等待 awaitable 对象完成后恢复执行async def task_a(): print("A start") await asyncio.sleep(1) print("A end") async def task_b(): print("B start") await asyncio.sleep(1) print("B end") async def main(): await task_a() await task_b() asyncio.run(main())三、并发执行3.1 asyncio.gatherasync def main(): await asyncio.gather(task_a(), task_b())3.2 asyncio.create_taskasync def main(): task1 = asyncio.create_task(task_a()) task2 = asyncio.create_task(task_b()) print("doing other work") await task1 await task23.3 超时控制async def slow_operation(): await asyncio.sleep(10) return "done" async def main(): try: result = await asyncio.wait_for(slow_operation(), timeout=3.0) print(result) except asyncio.TimeoutError: print("操作超时!") asyncio.run(main())3.4 asyncio.as_completedasync def fetch(simulated_time, name): await asyncio.sleep(simulated_time) return f"{name}: {simulated_time}s" async def main(): tasks = [ fetch(3, "task1"), fetch(1, "task2"), fetch(2, "task3"), ] for coro in asyncio.as_completed(tasks): result = await coro print(result) asyncio.run(main()) # 输出顺序: task2, task3, task1四、实际应用：并发爬虫import asyncio import aiohttp async def fetch_page(session, url): 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), } async def crawl(urls, concurrency=5): semaphore = asyncio.Semaphore(concurrency) async def limited_fetch(session, url): async with semaphore: return await fetch_page(session, url) async with aiohttp.ClientSession() as session: tasks = [limited_fetch(session, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True) urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(20)] results = asyncio.run(crawl(urls, concurrency=5)) for r in results: if isinstance(r, Exception): print(f"Error: {r}") else: print(f"{r[\"status\"]} - {r[\"url\"]} ({r[\"length\"]} bytes)")五、异步上下文管理器class AsyncDBConnection: async def __aenter__(self): print("连接数据库...") await asyncio.sleep(0.5) return self async def __aexit__(self, exc_type, exc_val, exc_tb): print("关闭连接...") await asyncio.sleep(0.1) async def query(self, sql): await asyncio.sleep(0.1) return f"Result of: {sql}" async def main(): async with AsyncDBConnection() as db: result = await db.query("SELECT * FROM users") print(result) asyncio.run(main())六、常见陷阱6.1 在异步中调用同步阻塞代码# 错误：requests 是同步库，会阻塞事件循环 async def bad_fetch(url): return requests.get(url).text # 正确：使用 aiohttp async def good_fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text()6.2 忘记 awaitasync def main(): asyncio.sleep(1) # 不会执行！返回 coroutine 对象 await asyncio.sleep(1) # 正确6.3 在异步代码中使用 time.sleepimport time async def main(): time.sleep(1) # 阻塞事件循环！ await asyncio.sleep(1) # 正确：让出控制权七、总结asyncio 的核心价值在于：在单线程内通过事件循环实现高并发 IO，避免了多线程的锁问题和上下文切换开销。记住三个关键点：IO 操作必须用异步库（aiohttp 而非 requests）、用 gather/create_task 实现并发、用 Semaphore 控制并发量。在高 IO 场景（爬虫、API 调用、数据库查询）中，asyncio 能带来数量级的性能提升。

Go + RabbitMQ 构建高并发任务队列实战 前言Go 语言的 goroutine 并发模型配合 RabbitMQ 的可靠消息投递，是构建高并发任务队列的经典方案。本文将实现一个完整的 Go + RabbitMQ 任务队列系统，包含生产者、消费者、重试机制和优雅退出。一、项目结构go-rabbitmq-queue/ ├── go.mod ├── config/ │ └── config.go ├── producer/ │ └── main.go ├── consumer/ │ └── main.go └── rabbitmq/ └── connection.go二、安装依赖go mod init github.com/yourname/go-rabbitmq-queue go get github.com/rabbitmq/amqp091-go三、RabbitMQ 连接封装package rabbitmq import ( "log" "time" amqp "github.com/rabbitmq/amqp091-go" ) type Config struct { URL string Exchange string Queue string RoutingKey string PrefetchCount int } type RabbitMQ struct { conn *amqp.Connection channel *amqp.Channel config Config } func New(cfg Config) (*RabbitMQ, error) { var conn *amqp.Connection var err error for i := 0; i < 5; i++ { conn, err = amqp.Dial(cfg.URL) if err == nil { break } log.Printf("连接失败(%d/5): %v", i+1, err) time.Sleep(3 * time.Second) } if err != nil { return nil, err } ch, err := conn.Channel() if err != nil { return nil, err } err = ch.ExchangeDeclare( cfg.Exchange, "direct", true, false, false, false, nil, ) if err != nil { return nil, err } args := amqp.Table{ "x-message-ttl": int32(60000), "x-dead-letter-exchange": cfg.Exchange + ".dlx", } _, err = ch.QueueDeclare( cfg.Queue, true, false, false, false, args, ) if err != nil { return nil, err } err = ch.QueueBind( cfg.Queue, cfg.RoutingKey, cfg.Exchange, false, nil, ) if err != nil { return nil, err } ch.Qos(cfg.PrefetchCount, 0, false) return &RabbitMQ{conn: conn, channel: ch, config: cfg}, nil } func (r *RabbitMQ) Channel() *amqp.Channel { return r.channel } func (r *RabbitMQ) Close() { r.channel.Close() r.conn.Close() }四、生产者package main import ( "encoding/json" "fmt" "log" "time" "github.com/yourname/go-rabbitmq-queue/rabbitmq" amqp "github.com/rabbitmq/amqp091-go" ) type Task struct { ID string `json:"id"` Type string `json:"type"` Payload interface{} `json:"payload"` } func main() { cfg := rabbitmq.Config{ URL: "amqp://admin:admin123@localhost:5672/", Exchange: "task.exchange", Queue: "task.queue", RoutingKey: "task.process", PrefetchCount: 10, } mq, err := rabbitmq.New(cfg) if err != nil { log.Fatal(err) } defer mq.Close() for i := 0; i < 100; i++ { task := Task{ ID: fmt.Sprintf("task-%d", i), Type: "email", Payload: map[string]string{ "to": fmt.Sprintf("user%d@example.com", i), "subject": "通知邮件", }, } body, _ := json.Marshal(task) err = mq.Channel().Publish( cfg.Exchange, cfg.RoutingKey, false, false, amqp.Publishing{ DeliveryMode: amqp.Persistent, ContentType: "application/json", Body: body, Timestamp: time.Now(), }, ) if err != nil { log.Printf("发送失败: %v", err) continue } log.Printf("发送任务: %s", task.ID) } log.Println("所有任务发送完成") }五、消费者（多 Worker 并发）package main import ( "encoding/json" "fmt" "log" "os" "os/signal" "sync" "syscall" "time" "github.com/yourname/go-rabbitmq-queue/rabbitmq" amqp "github.com/rabbitmq/amqp091-go" ) type Task struct { ID string `json:"id"` Type string `json:"type"` Payload interface{} `json:"payload"` } func processTask(task Task) error { log.Printf("处理任务: %s, 类型: %s", task.ID, task.Type) time.Sleep(500 * time.Millisecond) if time.Now().Unix()%10 == 0 { return fmt.Errorf("模拟处理失败") } log.Printf("任务完成: %s", task.ID) return nil } func startWorker(id int, mq *rabbitmq.RabbitMQ, wg *sync.WaitGroup) { defer wg.Done() msgs, err := mq.Channel().Consume( "task.queue", fmt.Sprintf("worker-%d", id), false, false, false, false, nil, ) if err != nil { log.Printf("Worker %d 启动失败: %v", id, err) return } log.Printf("Worker %d 启动", id) for msg := range msgs { var task Task if err := json.Unmarshal(msg.Body, &task); err != nil { log.Printf("Worker %d 解析失败: %v", id, err) msg.Nack(false, false) continue } if err := processTask(task); err != nil { log.Printf("Worker %d 处理失败: %s -> %v", id, task.ID, err) retryCount := getRetryCount(msg) if retryCount < 3 { msg.Nack(false, true) } else { log.Printf("Worker %d 任务 %s 重试超限，进入死信", id, task.ID) msg.Nack(false, false) } continue } msg.Ack(false) } log.Printf("Worker %d 退出", id) } func getRetryCount(msg amqp.Delivery) int { if deaths, ok := msg.Headers["x-death"].([]interface{}); ok && len(deaths) > 0 { if death, ok := deaths[0].(amqp.Table); ok { if count, ok := death["count"].(int64); ok { return int(count) } } } return 0 } func main() { cfg := rabbitmq.Config{ URL: "amqp://admin:admin123@localhost:5672/", Exchange: "task.exchange", Queue: "task.queue", RoutingKey: "task.process", PrefetchCount: 5, } mq, err := rabbitmq.New(cfg) if err != nil { log.Fatal(err) } defer mq.Close() var wg sync.WaitGroup workerCount := 5 for i := 1; i <= workerCount; i++ { wg.Add(1) go startWorker(i, mq, &wg) } sigs := make(chan os.Signal, 1) signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) <-sigs log.Println("收到退出信号，等待 worker 完成...") mq.Close() wg.Wait() log.Println("所有 worker 已退出") }六、死信队列消费者func startDeadLetterConsumer(mq *rabbitmq.RabbitMQ) { ch := mq.Channel() ch.ExchangeDeclare("task.exchange.dlx", "fanout", true, false, false, false, nil) _, _ = ch.QueueDeclare("task.dlq", true, false, false, false, nil) _ = ch.QueueBind("task.dlq", "", "task.exchange.dlx", false, nil) msgs, _ := ch.Consume("task.dlq", "dlq-consumer", false, false, false, false, nil) go func() { for msg := range msgs { log.Printf("死信消息: %s", string(msg.Body)) msg.Ack(false) } }() }七、架构总结Producer → [task.exchange] → [task.queue] → 5个 Worker 并发消费 ↓ (失败/Nack) [task.exchange.dlx] → [task.dlq] → 死信消费者记录总结Go + RabbitMQ 的组合充分发挥了各自优势：Go 的 goroutine 让消费者可以轻松开几十个并发 worker，RabbitMQ 的 ACK 机制保证消息不丢失。生产环境注意：设置合理的 prefetch 避免消息堆积在单个 worker、实现死信队列处理失败消息、添加优雅退出逻辑避免消息中断。

Nginx 反向代理配置详解与性能优化实践 前言Nginx 作为高性能 HTTP 服务器和反向代理，在生产环境中应用极为广泛。本文将系统讲解 Nginx 反向代理的核心配置，并分享实际项目中的性能优化经验。一、反向代理基础配置1.1 基本反向代理server { listen 80; server_name example.com; location / { proxy_pass http://127.0.0.1:3000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }1.2 负载均衡配置upstream backend { server 192.168.1.10:3000 weight=3; server 192.168.1.11:3000 weight=2; server 192.168.1.12:3000 weight=1; keepalive 32; } server { listen 80; location / { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; } }二、性能优化要点2.1 Worker 进程配置worker_processes auto; worker_cpu_affinity auto; events { worker_connections 10240; use epoll; multi_accept on; }2.2 Gzip 压缩gzip on; gzip_min_length 1k; gzip_comp_level 6; gzip_types text/plain text/css application/json application/javascript text/xml application/xml text/javascript image/svg+xml; gzip_vary on; gzip_proxied any;2.3 静态文件缓存location ~* \.(jpg|jpeg|png|gif|ico|css|js|woff2)$ { expires 30d; add_header Cache-Control "public, immutable"; access_log off; }三、SSL/TLS 配置server { listen 443 ssl http2; server_name example.com; ssl_certificate /etc/nginx/ssl/fullchain.pem; ssl_certificate_key /etc/nginx/ssl/privkey.pem; ssl_protocols TLSv1.2 TLSv1.3; ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256; ssl_prefer_server_ciphers on; ssl_session_cache shared:SSL:10m; ssl_session_timeout 1d; add_header Strict-Transport-Security "max-age=63072000" always; }四、常见问题排查4.1 502 Bad Gateway通常原因：后端服务未启动、端口不通、网络问题。# 检查后端服务 curl -I http://127.0.0.1:3000 # 查看错误日志 tail -f /var/log/nginx/error.log # 检查端口 ss -tlnp | grep 30004.2 504 Gateway Timeout后端响应超时，适当调大 proxy_read_timeout。五、总结Nginx 反向代理配置看似简单，但要在生产环境中跑出最佳性能，需要在 worker 模型、连接复用、压缩缓存、SSL 优化等多个维度精细调优。建议每次修改配置后用 nginx -t 检查语法，逐步调整观察效果。

Docker Compose 多容器编排实战指南 前言Docker Compose 是定义和运行多容器 Docker 应用的工具。本文通过一个完整的 Web 应用示例，讲解 Compose 的核心用法和最佳实践。一、项目结构myapp/ ├── docker-compose.yml ├── nginx/ │ └── default.conf ├── php/ │ └── Dockerfile ├── src/ │ └── index.php └── mysql/ └── init.sql二、完整的 LNMP 编排version: "3.8" services: nginx: image: nginx:1.25-alpine ports: - "80:80" - "443:443" volumes: - ./src:/var/www/html - ./nginx/default.conf:/etc/nginx/conf.d/default.conf depends_on: - php restart: unless-stopped networks: - app-network php: build: context: ./php dockerfile: Dockerfile volumes: - ./src:/var/www/html environment: - DB_HOST=mysql - DB_NAME=myapp - DB_USER=root - DB_PASS=secret networks: - app-network mysql: image: mysql:8.0 ports: - "3306:3306" volumes: - mysql-data:/var/lib/mysql - ./mysql/init.sql:/docker-entrypoint-initdb.d/init.sql environment: MYSQL_ROOT_PASSWORD: secret MYSQL_DATABASE: myapp networks: - app-network redis: image: redis:7-alpine ports: - "6379:6379" volumes: - redis-data:/data networks: - app-network volumes: mysql-data: redis-data: networks: app-network: driver: bridge三、常用命令# 启动所有服务 docker-compose up -d # 查看运行状态 docker-compose ps # 查看日志 docker-compose logs -f nginx # 重新构建 docker-compose build --no-cache php # 停止并删除容器 docker-compose down # 停止并删除容器和数据卷 docker-compose down -v四、进阶技巧4.1 健康检查mysql: image: mysql:8.0 healthcheck: test: ["CMD", "mysqladmin", "ping", "-h", "localhost"] interval: 10s timeout: 5s retries: 3 start_period: 30s4.2 资源限制php: build: ./php deploy: resources: limits: cpus: "2" memory: 512M reservations: memory: 256M4.3 按需启动依赖php: depends_on: mysql: condition: service_healthy redis: condition: service_started五、总结Docker Compose 让多容器应用的定义、运行和管理变得简单。通过合理的目录结构、网络隔离、数据卷管理和健康检查，可以构建出生产级的容器编排方案。

Vue3 Composition API 深度解析：从 ref 到自定义 Hook 前言Vue3 的 Composition API 是对 Options API 的范式升级。它解决了大型组件中逻辑复用和组织的问题，让代码更灵活、更可维护。一、响应式基础1.1 ref vs reactiveimport { ref, reactive, toRefs } from "vue"; // ref：适用于基本类型 const count = ref(0); count.value++; // reactive：适用于对象 const state = reactive({ name: "张三", age: 25, }); // reactive 的陷阱：解构会失去响应性 const { name } = state; // name 不再是响应式 // 解决方案：toRefs const { name: reactiveName } = toRefs(state);1.2 computedimport { ref, computed } from "vue"; const firstName = ref("张"); const lastName = ref("三"); // 只读计算属性 const fullName = computed(() => `${firstName.value}${lastName.value}`); // 可写计算属性 const fullNameWritable = computed({ get() { return `${firstName.value}${lastName.value}`; }, set(newValue: string) { firstName.value = newValue[0]; lastName.value = newValue.slice(1); }, });二、watch 和 watchEffect2.1 watchimport { ref, watch } from "vue"; const keyword = ref(""); // 监听单个 ref watch(keyword, (newVal, oldVal) => { console.log(`关键词从 ${oldVal} 变为 ${newVal}`); }); // 监听多个源 watch([keyword, category], ([newKeyword, newCategory]) => { search(newKeyword, newCategory); }); // 深度监听对象 watch( () => state, (newState) => { saveToStorage(newState); }, { deep: true } );2.2 watchEffectimport { ref, watchEffect } from "vue"; const userId = ref(1); // 自动追踪依赖，立即执行 watchEffect(() => { fetchUser(userId.value); // userId 变化时自动重新执行 });三、自定义 Hook（组合式函数）3.1 useMousePosition// composables/useMousePosition.ts import { ref, onMounted, onUnmounted } from "vue"; export function useMousePosition() { const x = ref(0); const y = ref(0); function update(e: MouseEvent) { x.value = e.pageX; y.value = e.pageY; } onMounted(() => { window.addEventListener("mousemove", update); }); onUnmounted(() => { window.removeEventListener("mousemove", update); }); return { x, y }; }使用：<script setup lang="ts"> import { useMousePosition } from "@/composables/useMousePosition"; const { x, y } = useMousePosition(); </script> <template> <p>鼠标位置: {{ x }}, {{ y }}</p> </template>3.2 useFetch// composables/useFetch.ts import { ref, watchEffect } from "vue"; export function useFetch<T>(url: () => string) { const data = ref<T | null>(null); const error = ref<string | null>(null); const loading = ref(false); watchEffect(async () => { loading.value = true; error.value = null; try { const res = await fetch(url()); if (!res.ok) throw new Error(`HTTP ${res.status}`); data.value = await res.json() as T; } catch (e) { error.value = (e as Error).message; } finally { loading.value = false; } }); return { data, error, loading }; }四、生命周期钩子import { onMounted, onUpdated, onUnmounted, onBeforeMount, onBeforeUpdate, onBeforeUnmount } from "vue"; onBeforeMount(() => { /* 组件挂载前 */ }); onMounted(() => { /* DOM 已挂载 */ }); onBeforeUpdate(() => { /* 响应式数据变更，DOM 更新前 */ }); onUpdated(() => { /* DOM 更新后 */ }); onBeforeUnmount(() => { /* 组件卸载前 */ }); onUnmounted(() => { /* 组件已卸载 */ });五、总结Composition API 的核心优势在于逻辑复用和代码组织。通过自定义 Hook，可以将组件逻辑提取为可复用的函数，告别 Options API 时代 mixin 的命名冲突和数据来源不透明问题。建议在项目中优先使用 <script setup> 语法，配合 TypeScript 获得最佳开发体验。