RabbitMQ 消息队列入门与实战：从原理到 Spring Boot 整合 前言RabbitMQ 是基于 AMQP 协议的开源消息中间件，广泛应用于微服务架构中的异步解耦、削峰填谷和可靠投递场景。本文将从核心概念讲起，最终完成一个 Spring Boot 整合 RabbitMQ 的实战项目。一、核心概念1.1 AMQP 模型Producer → Exchange → (Binding) → Queue → Consumer概念说明Producer消息生产者Exchange交换机，接收消息并路由到队列Queue消息队列，存储消息Binding交换机与队列之间的绑定关系Routing Key路由键，Exchange 根据它决定消息去向Consumer消息消费者1.2 四种交换机类型类型说明适用场景Direct精确匹配 Routing Key点对点定向投递Fanout广播到所有绑定队列广播通知Topic模式匹配 Routing Key按主题订阅Headers根据消息头匹配复杂路由条件二、Docker 快速部署docker run -d --name rabbitmq \ -p 5672:5672 \ -p 15672:15672 \ -e RABBITMQ_DEFAULT_USER=admin \ -e RABBITMQ_DEFAULT_PASS=admin123 \ rabbitmq:3.13-management5672：AMQP 通信端口15672：管理界面端口，浏览器访问 http://localhost:15672三、Spring Boot 整合3.1 添加依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>3.2 配置spring: rabbitmq: host: localhost port: 5672 username: admin password: admin123 virtual-host: /3.3 队列与交换机配置@Configuration public class RabbitMQConfig { public static final String EXCHANGE = "order.exchange"; public static final String QUEUE = "order.queue"; public static final String ROUTING_KEY = "order.create"; @Bean public TopicExchange orderExchange() { return new TopicExchange(EXCHANGE); } @Bean public Queue orderQueue() { return QueueBuilder.durable(QUEUE) .withArgument("x-message-ttl", 60000) .withArgument("x-dead-letter-exchange", "order.dlx") .build(); } @Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(ROUTING_KEY); } }3.4 生产者@Service public class OrderProducer { @Autowired private RabbitTemplate rabbitTemplate; public void sendOrder(OrderDTO order) { rabbitTemplate.convertAndSend( RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTING_KEY, order, message -> { message.getMessageProperties() .setDeliveryMode(MessageDeliveryMode.PERSISTENT); return message; } ); } }3.5 消费者@Component public class OrderConsumer { @RabbitListener(queues = RabbitMQConfig.QUEUE) public void handleOrder(OrderDTO order, Channel channel, Message message) throws IOException { try { processOrder(order); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { channel.basicNack( message.getMessageProperties().getDeliveryTag(), false, true ); } } private void processOrder(OrderDTO order) { System.out.println("处理订单: " + order.getId()); } }3.6 配置手动 ACKspring: rabbitmq: listener: simple: acknowledge-mode: manual prefetch: 10 retry: enabled: true max-attempts: 3四、可靠性投递方案4.1 生产者确认（Publisher Confirm）@Configuration public class RabbitConfig { @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory cf) { RabbitTemplate template = new RabbitTemplate(cf); template.setConfirmCallback((correlationData, ack, cause) -> { if (!ack) { log.error("消息未到达Exchange: {}", cause); } }); template.setReturnsCallback(returned -> { log.error("消息未路由到Queue: {}", returned.getMessage()); }); template.setMandatory(true); return template; } }4.2 死信队列（DLX）消息变成死信的条件：消息被消费者拒绝（basicNack + requeue=false）消息 TTL 过期队列达到最大长度@Bean public Queue deadLetterQueue() { return QueueBuilder.durable("order.dlq").build(); } @Bean public FanoutExchange deadLetterExchange() { return new FanoutExchange("order.dlx"); } @Bean public Binding dlxBinding() { return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()); }五、常见问题问题原因解决方案消息丢失消费者宕机未 ACK手动 ACK + 持久化消息重复网络抖动导致重复投递消费端幂等性（唯一ID去重）消息堆积消费速度小于生产速度增加消费者 / 限流队列阻塞单条消息处理过慢合理设置 prefetch总结RabbitMQ 在微服务架构中承担着异步解耦和流量削峰的关键角色。生产环境中务必做到：持久化 + 手动 ACK + 死信队列 + 消费幂等，才能保证消息的可靠投递。

Go 语言基础教程：从语法到并发编程 前言Go 语言以简洁的语法、出色的并发模型和编译速度著称，在云原生和服务端开发领域占据重要地位。本文将系统介绍 Go 的核心语法和并发编程。一、环境与项目1.1 安装go version # go1.22.x # 设置国内代理 go env -w GOPROXY=https://goproxy.cn,direct go env -w GO111MODULE=on1.2 创建项目mkdir myapp && cd myapp go mod init github.com/yourname/myapp二、基础语法2.1 变量与类型package main import "fmt" func main() { var name string = "Go" var age int = 18 var score = 95.5 isReady := true x, y := 10, 20 fmt.Println(name, age, score, isReady, x, y) }2.2 结构体与方法type User struct { ID int Name string Age int } func (u User) Greet() string { return fmt.Sprintf("Hi, I am %s", u.Name) } func (u *User) SetAge(age int) { u.Age = age } func main() { u := User{ID: 1, Name: "张三", Age: 25} fmt.Println(u.Greet()) u.SetAge(26) fmt.Println(u.Age) }2.3 接口type Animal interface { Speak() string } type Dog struct{ Name string } type Cat struct{ Name string } func (d Dog) Speak() string { return d.Name + ": 汪汪!" } func (c Cat) Speak() string { return c.Name + ": 喵喵!" } func makeSpeak(a Animal) { fmt.Println(a.Speak()) } func main() { makeSpeak(Dog{"旺财"}) makeSpeak(Cat{"咪咪"}) }2.4 错误处理func divide(a, b float64) (float64, error) { if b == 0 { return 0, fmt.Errorf("division by zero") } return a / b, nil } func main() { result, err := divide(10, 0) if err != nil { fmt.Println("Error:", err) return } fmt.Println("Result:", result) }三、并发编程3.1 Goroutinefunc sayHello(name string) { fmt.Printf("Hello, %s!\n", name) } func main() { go sayHello("张三") go sayHello("李四") time.Sleep(time.Second) }3.2 Channelfunc main() { ch := make(chan string) go func() { ch <- "来自 goroutine 的消息" }() msg := <-ch fmt.Println(msg) }3.3 带缓冲的 Channelch := make(chan int, 3) ch <- 1 ch <- 2 ch <- 3 // ch <- 4 // 阻塞！缓冲区已满 fmt.Println(<-ch) // 13.4 select 多路复用func main() { ch1 := make(chan string) ch2 := make(chan string) go func() { time.Sleep(1 * time.Second) ch1 <- "来自 ch1" }() go func() { time.Sleep(2 * time.Second) ch2 <- "来自 ch2" }() select { case msg := <-ch1: fmt.Println(msg) case msg := <-ch2: fmt.Println(msg) } }3.5 WaitGroup 等待多个 goroutinefunc worker(id int, wg *sync.WaitGroup) { defer wg.Done() fmt.Printf("Worker %d started\n", id) time.Sleep(time.Second) fmt.Printf("Worker %d done\n", id) } func main() { var wg sync.WaitGroup for i := 1; i <= 5; i++ { wg.Add(1) go worker(i, &wg) } wg.Wait() fmt.Println("All workers done") }3.6 并发安全：Mutextype Counter struct { mu sync.Mutex count int } func (c *Counter) Increment() { c.mu.Lock() defer c.mu.Unlock() c.count++ } func (c *Counter) Value() int { c.mu.Lock() defer c.mu.Unlock() return c.count } func main() { c := &Counter{} var wg sync.WaitGroup for i := 0; i < 1000; i++ { wg.Add(1) go func() { defer wg.Done() c.Increment() }() } wg.Wait() fmt.Println("Final count:", c.Value()) }四、HTTP 服务func main() { http.HandleFunc("/api/hello", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]string{ "msg": "Hello from Go!", }) }) fmt.Println("Server running on :8080") log.Fatal(http.ListenAndServe(":8080", nil)) }五、总结Go 的核心优势在于：语法简洁上手快、goroutine 轻量级并发、编译速度快、标准库强大。掌握 struct/interface/goroutine/channel 这四个核心概念，就能写出高质量的 Go 程序。建议在实际项目中配合 Gin/Echo 等 Web 框架和 GORM 等 ORM 框架，进一步提升开发效率。

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、实现死信队列处理失败消息、添加优雅退出逻辑避免消息中断。

Node.js JWT 认证完整实现：注册登录到权限控制 前言JWT（JSON Web Token）是现代 Web 应用中最流行的认证方案之一。本文将使用 Node.js + Express 实现一个完整的 JWT 认证系统，包含注册、登录、Token 刷新和权限控制。一、项目初始化mkdir jwt-auth && cd jwt-auth npm init -y npm install express jsonwebtoken bcryptjs cors dotenv npm install -D typescript @types/express @types/node ts-node-dev二、项目结构src/ ├── config/ │ └── index.ts ├── middleware/ │ └── auth.ts ├── routes/ │ └── auth.ts ├── utils/ │ └── jwt.ts ├── types/ │ └── index.ts └── app.ts三、核心实现3.1 JWT 工具类// src/utils/jwt.ts import jwt from "jsonwebtoken"; import { config } from "../config"; export interface JwtPayload { userId: string; role: string; } export function signAccessToken(payload: JwtPayload): string { return jwt.sign(payload, config.JWT_SECRET, { expiresIn: "15m", }); } export function signRefreshToken(payload: JwtPayload): string { return jwt.sign(payload, config.JWT_REFRESH_SECRET, { expiresIn: "7d", }); } export function verifyAccessToken(token: string): JwtPayload { return jwt.verify(token, config.JWT_SECRET) as JwtPayload; } export function verifyRefreshToken(token: string): JwtPayload { return jwt.verify(token, config.JWT_REFRESH_SECRET) as JwtPayload; }3.2 认证中间件// src/middleware/auth.ts import { Request, Response, NextFunction } from "express"; import { verifyAccessToken } from "../utils/jwt"; // 扩展 Request 类型 declare global { namespace Express { interface Request { user?: { userId: string; role: string }; } } } // 基本认证 export function authRequired(req: Request, res: Response, next: NextFunction) { const authHeader = req.headers.authorization; if (!authHeader?.startsWith("Bearer ")) { return res.status(401).json({ code: 401, msg: "未提供认证令牌" }); } const token = authHeader.substring(7); try { const payload = verifyAccessToken(token); req.user = { userId: payload.userId, role: payload.role }; next(); } catch { return res.status(401).json({ code: 401, msg: "令牌无效或已过期" }); } } // 角色权限控制 export function requireRole(...roles: string[]) { return (req: Request, res: Response, next: NextFunction) => { if (!req.user) { return res.status(401).json({ code: 401, msg: "未认证" }); } if (!roles.includes(req.user.role)) { return res.status(403).json({ code: 403, msg: "权限不足" }); } next(); }; }3.3 路由实现// src/routes/auth.ts import { Router, Response } from "express"; import bcrypt from "bcryptjs"; import { authRequired, requireRole } from "../middleware/auth"; import { signAccessToken, signRefreshToken, verifyRefreshToken } from "../utils/jwt"; const router = Router(); // 模拟用户存储（实际应使用数据库） const users: Map<string, { id: string; username: string; password: string; role: string }> = new Map(); const refreshTokens: Set<string> = new Set(); // 注册 router.post("/register", async (req, res: Response) => { const { username, password } = req.body; if (!username || !password) { return res.status(400).json({ code: 400, msg: "用户名和密码不能为空" }); } if (users.has(username)) { return res.status(409).json({ code: 409, msg: "用户已存在" }); } const hashedPassword = await bcrypt.hash(password, 10); const id = Date.now().toString(); users.set(username, { id, username, password: hashedPassword, role: "user" }); res.json({ code: 0, msg: "注册成功" }); }); // 登录 router.post("/login", async (req, res: Response) => { const { username, password } = req.body; const user = users.get(username); if (!user) { return res.status(404).json({ code: 404, msg: "用户不存在" }); } const valid = await bcrypt.compare(password, user.password); if (!valid) { return res.status(401).json({ code: 401, msg: "密码错误" }); } const payload = { userId: user.id, role: user.role }; const accessToken = signAccessToken(payload); const refreshToken = signRefreshToken(payload); refreshTokens.add(refreshToken); res.json({ code: 0, msg: "登录成功", data: { accessToken, refreshToken, user: { id: user.id, username: user.username, role: user.role } }, }); }); // 刷新 Token router.post("/refresh", (req, res: Response) => { const { refreshToken } = req.body; if (!refreshToken || !refreshTokens.has(refreshToken)) { return res.status(401).json({ code: 401, msg: "无效的刷新令牌" }); } try { const payload = verifyRefreshToken(refreshToken); refreshTokens.delete(refreshToken); const newPayload = { userId: payload.userId, role: payload.role }; const newAccessToken = signAccessToken(newPayload); const newRefreshToken = signRefreshToken(newPayload); refreshTokens.add(newRefreshToken); res.json({ code: 0, msg: "刷新成功", data: { accessToken: newAccessToken, refreshToken: newRefreshToken }, }); } catch { refreshTokens.delete(refreshToken); return res.status(401).json({ code: 401, msg: "刷新令牌已过期" }); } }); // 登出 router.post("/logout", (req, res: Response) => { const { refreshToken } = req.body; refreshTokens.delete(refreshToken); res.json({ code: 0, msg: "已登出" }); }); // 受保护的路由 router.get("/profile", authRequired, (req, res: Response) => { res.json({ code: 0, data: { userId: req.user!.userId, role: req.user!.role } }); }); // 管理员路由 router.get("/admin/users", authRequired, requireRole("admin"), (req, res: Response) => { const userList = Array.from(users.values()).map(u => ({ id: u.id, username: u.username, role: u.role, })); res.json({ code: 0, data: userList }); }); export default router;3.4 主应用// src/app.ts import express from "express"; import cors from "cors"; import dotenv from "dotenv"; import authRoutes from "./routes/auth"; dotenv.config(); const app = express(); app.use(cors()); app.use(express.json()); app.use("/api/auth", authRoutes); app.get("/", (_req, res) => { res.json({ msg: "JWT Auth API" }); }); app.listen(3000, () => { console.log("Server running on http://localhost:3000"); });四、安全最佳实践Access Token 短时效（15 分钟），Refresh Token 长时效（7 天）Refresh Token 一次性使用：每次刷新后旧 Token 失效HTTPS 传输：生产环境必须使用 HTTPS密码加密：使用 bcrypt，不要用 MD5/SHA1环境变量管理密钥：不要硬编码到代码中Token 黑名单：登出时将 Token 加入黑名单（Redis）总结JWT 认证方案无状态、易扩展，适合前后端分离架构。通过 Access Token + Refresh Token 双 Token 机制，可以在安全性和用户体验之间取得平衡。生产环境中建议将 Refresh Token 存储在 Redis 中，方便统一管理和吊销。