内部邮件系统保姆级教程：面试被问原理？3步搞定核心逻辑
面试被问“内部邮件系统怎么设计”，你答不上来？别慌，这不是背诵题，是考察你对分布式系统、状态机和高并发处理的实战理解。很多人只会调API，一到追问“如何保证消息不丢”、“如何防重放”就卡壳。这篇保姆级教程，不讲虚的，直接带你从零搭建一个能跑、能测、能扛住基本业务量的内部邮件服务。我们不用复杂的大框架，就用Python + FastAPI + Redis + PostgreSQL，把最核心的发送、接收、已读标记、防重逻辑全部拆解清楚。你跟着敲一遍，面试时再遇到类似问题，就能自信地说：“我做过，我知道坑在哪。”
项目目标与核心难点拆解
在动手写代码前，先明确我们要解决什么。内部邮件系统不是简单的“发邮件”，它有几个硬性指标：可靠性（消息必须送达或明确失败）、一致性（用户看到的已读状态必须实时准确）、防重（网络抖动导致的重复请求不能产生两封邮件）。很多初学者容易忽略“防重”和“已读状态的并发更新”，导致线上出现“同一封邮件收到两次”或“明明已读却显示未读”的Bug。
我们的目标很清晰：实现一个最小可用版本（MVP），包含以下四个核心功能：发送接口：支持单发和多发，返回唯一的事务ID。
接收与查询接口：用户只能查自己的收件箱，支持分页。
已读标记接口：用户标记某邮件为已读，需处理并发竞争。
幂等性控制：客户端携带Client-Request-ID，服务端确保同一ID只处理一次。为什么强调幂等性？因为在分布式环境下，前端可能因为超时重试，或者网络层代理转发，导致同一个请求到达后端多次。如果没有幂等控制，用户就会收到重复邮件，这是内部系统的大忌。
目录结构设计
一个清晰的项目结构能让你在面试时快速定位代码，也方便后续扩展。我们采用分层架构，职责分离。
internal-mail-service/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 入口
│ ├── config.py # 配置管理
│ ├── models/ # 数据库模型
│ │ ├── __init__.py
│ │ ├── email.py # Email 表结构
│ │ └── user.py # User 表结构
│ ├── schemas/ # Pydantic 请求/响应模型
│ │ ├── __init__.py
│ │ └── email.py # 数据校验与序列化
│ ├── services/ # 业务逻辑层
│ │ ├── __init__.py
│ │ ├── email_service.py # 核心邮件逻辑
│ │ └── idempotency.py # 幂等性服务
│ ├── api/ # API 路由层
│ │ ├── __init__.py
│ │ └── v1/
│ │ └── emails.py
│ └── utils/ # 工具函数
│ ├── __init__.py
│ └── redis_client.py
├── alembic/ # 数据库迁移
├── tests/ # 单元测试
├── requirements.txt
└── .env关键说明：services/ 层是核心，所有业务逻辑都在这里，API层只做参数校验和调用。
idempotency.py 单独抽离，因为幂等性是一个通用能力，未来其他接口（如转账、下单）也能复用。
使用 alembic 管理数据库迁移，避免直接改表结构导致线上事故。核心代码实现
1. 数据库模型定义
先定义数据结构。邮件表需要记录发送者、接收者、主题、内容、状态（未读/已读/删除）以及时间戳。
# app/models/email.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
from datetime import datetimeclass Email(Base):__tablename__ = emailsid = Column(Integer, primary_key=True, index=True)# 幂等键，用于防重，唯一索引client_request_id = Column(String(128), unique=True, index=True, nullable=False)sender_id = Column(Integer, ForeignKey(users.id), nullable=False)recipient_id = Column(Integer, ForeignKey(users.id), nullable=False, index=True)subject = Column(String(255), nullable=False)content = Column(Text, nullable=False)# 状态：0=未读, 1=已读, 2=删除status = Column(Integer, default=0, index=True)created_at = Column(DateTime, default=datetime.utcnow)read_at = Column(DateTime, nullable=True)# 关联关系，方便查询sender = relationship(User, foreign_keys=[sender_id])recipient = relationship(User, foreign_keys=[recipient_id])注意：client_request_id 加了 unique=True 和 index=True，这是数据库层面的最后一道防重防线。即使Redis失效，数据库也不会插入重复数据。
2. 幂等性服务实现
幂等性的核心思想是：先查，后写，用锁保护。我们使用Redis的 SETNX（Set if Not Exists）命令来实现原子性的“检查并设置”。
# app/services/idempotency.py
import redis
import json
from app.config import settings
from typing import Optionalclass IdempotencyService:def __init__(self):# 初始化 Redis 连接池self.redis_client = redis.Redis(host=settings.REDIS_HOST,port=settings.REDIS_PORT,db=0,decode_responses=True)def check_and_set(self, client_request_id: str, result: dict) - bool:检查是否已处理过该请求。如果未处理，则设置键值并返回True，表示允许继续执行。如果已处理，返回False，并返回之前的结果。key = fidempotency:{client_request_id}# 设置过期时间为24小时，避免Redis数据无限膨胀try:# SETNX: 如果键不存在，则设置并返回1；如果存在，返回0if self.redis_client.setnx(key, json.dumps(result), ex=86400):return Trueelse:# 键已存在，说明重复请求return Falseexcept redis.RedisError as e:# Redis 故障时，降级为数据库唯一约束兜底# 这里可以选择抛异常或记录日志后继续执行print(fRedis error: {e})return Truedef get_previous_result(self, client_request_id: str) - Optional[dict]:获取之前处理的结果key = fidempotency:{client_request_id}data = self.redis_client.get(key)if data:return json.loads(data)return None为什么用 setnx 而不是 get + set？
因为 get 和 set 是两步操作，在高并发下存在竞态条件：两个请求同时 get 发现为空，然后同时 set，导致都以为自己是第一次。setnx 是原子操作，Redis保证只有一个请求能成功设置，其他请求会立即得到失败信号。
3. 邮件发送核心逻辑
这是最复杂的部分，需要结合数据库事务和幂等性。
# app/services/email_service.py
from sqlalchemy.orm import Session
from app.models.email import Email
from app.schemas.email import EmailCreate, EmailResponse
from app.services.idempotency import IdempotencyService
from fastapi import HTTPException, status
from typing import List
import uuidclass EmailService:def __init__(self, db: Session, idempotency_service: IdempotencyService):self.db = dbself.idempotency = idempotency_servicedef send_email(self, sender_id: int, email_in: EmailCreate) - EmailResponse:发送邮件1. 幂等性检查2. 业务校验3. 数据库写入4. 记录幂等结果# 1. 幂等性检查# 如果之前处理过，直接返回之前的结果prev_result = self.idempotency.get_previous_result(email_in.client_request_id)if prev_result:return EmailResponse(**prev_result)# 检查是否允许写入（原子操作）if not self.idempotency.check_and_set(email_in.client_request_id, {}):# 理论上不会走到这里，因为上面已经获取了结果# 但如果 Redis 和 DB 状态不一致，这里兜底raise HTTPException(status_code=409, detail=Duplicate request)try:# 2. 业务校验# 检查接收者是否存在（实际项目中可能支持多收件人，这里简化为单收件人）if not self.db.query(User).filter(User.id == email_in.recipient_id).first():raise HTTPException(status_code=404, detail=Recipient not found)# 3. 创建邮件对象email_obj = Email(client_request_id=email_in.client_request_id,sender_id=sender_id,recipient_id=email_in.recipient_id,subject=email_in.subject,content=email_in.content,status=0 # 默认未读)self.db.add(email_obj)self.db.commit()self.db.refresh(email_obj)# 4. 记录幂等结果，供后续重试返回result = {id: email_obj.id,client_request_id: email_in.client_request_id,status: sent}self.idempotency.set_result(email_in.client_request_id, result)return EmailResponse(id=email_obj.id,subject=email_obj.subject,status=sent)except Exception as e:# 发生异常，回滚事务，并删除幂等键，允许客户端重试self.db.rollback()self.idempotency.delete_key(email_in.client_request_id)raise HTTPException(status_code=500, detail=fFailed to send email: {str(e)})关键细节解析：异常处理：如果数据库写入失败，必须回滚事务，并且删除Redis中的幂等键。否则，客户端重试时会被幂等性拦截，永远无法成功。
结果缓存：成功后，将结果存入Redis。如果客户端因为网络原因没收到响应而重试，服务端直接返回缓存的结果，无需再查数据库，性能极高。4. API路由层
# app/api/v1/emails.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.services.email_service import EmailService
from app.services.idempotency import IdempotencyService
from app.schemas.email import EmailCreate, EmailResponserouter = APIRouter(prefix=/emails, tags=[emails])@router.post(/, response_model=EmailResponse)
def send_email(email_in: EmailCreate,current_user_id: int = Depends(get_current_user), # 假设从Token解析db: Session = Depends(get_db)
):发送邮件接口请求体必须包含 client_request_ididempotency_service = IdempotencyService()email_service = EmailService(db, idempotency_service)return email_service.send_email(current_user_id, email_in)@router.get(/unread, response_model=List[EmailResponse])
def get_unread_emails(skip: int = 0,limit: int = 10,current_user_id: int = Depends(get_current_user),db: Session = Depends(get_db)
):获取未读邮件，支持分页emails = db.query(Email).filter(Email.recipient_id == current_user_id,Email.status == 0).offset(skip).limit(limit).all()return [EmailResponse(id=e.id, subject=e.subject, status=unread) for e in emails]运行与测试
环境准备
安装依赖：
pip install fastapi uvicorn sqlalchemy psycopg2-binary redis pydantic alembic python-dotenv配置 .env 文件：
DATABASE_URL=postgresql://user:pass@localhost:5432/mail_db
REDIS_HOST=localhost
REDIS_PORT=6379启动服务：
uvicorn app.main:app --reload测试幂等性
使用 curl 模拟重复请求：
# 第一次请求
curl -X POST http://localhost:8000/emails/ \-H Content-Type: application/json \-d '{client_request_id: test-123,recipient_id: 1,subject: Hello,content: World}'
# 响应: {id: 1, subject: Hello, status: sent}# 第二次请求，完全相同的 client_request_id
curl -X POST http://localhost:8000/emails/ \-H Content-Type: application/json \-d '{client_request_id: test-123,recipient_id: 1,subject: Hello,content: World}'
# 响应: {id: 1, subject: Hello, status: sent} (ID相同，证明防重成功)测试要点：检查数据库 emails 表，client_request_id 为 test-123 的记录只有一条。
检查Redis，idempotency:test-123 键存在，且值为JSON字符串。
模拟数据库写入失败（如手动停掉PostgreSQL），再发送请求，确认Redis键被删除，恢复数据库后重试能成功。优化扩展与避坑指南
1. 高并发下的已读状态更新
如果多个用户同时标记同一封邮件为已读（虽然内部系统少见，但技术上要防范），直接使用 UPDATE ... SET status=1 会有竞态条件。建议使用乐观锁或Redis位图。
方案A：乐观锁（推荐用于小规模）
在 Email 表中增加 version 字段。
UPDATE emails SET status=1, version=version+1 WHERE id=123 AND version=5;如果影响行数为0，说明被其他请求更新，重试或报错。
方案B：Redis位图（适合海量数据）
用Redis的 SETBIT 命令标记已读。
# key: user:123:read_bits
# bit position: email_id
redis_client.setbit(user:123:read_bits, email_id, 1)查询未读时，结合数据库和Redis位图判断。这种方案性能极高，但需要处理Redis和数据库的一致性。
2. 消息可靠性保障
目前我们的实现是同步写入数据库，如果数据库挂了，请求会失败。对于更高要求，可以引入消息队列（如Kafka、RabbitMQ）：请求先到MQ，返回成功。
消费者从MQ拉取消息，写入数据库。
如果写入失败，消息重新入队，直到成功或进入死信队列。但注意：MQ会引入延迟，且增加系统复杂度。内部系统通常对实时性要求高，同步写入+幂等性已足够。
3. 避免常见坑不要只用Redis防重：Redis可能宕机或数据丢失，必须配合数据库唯一约束。
幂等键要足够唯一：建议使用 UUID + 用户ID + 业务类型 组合，避免跨用户冲突。
超时设置：Redis操作要设置超时，避免网络抖动导致接口卡死。
日志记录：在幂等性检查失败时，记录警告日志，方便排查是否是客户端Bug。小结
通过这个保姆级教程，我们搭建了一个具备生产级基础能力的内部邮件系统。核心不是代码本身，而是幂等性设计和状态一致性处理的思路。面试时，如果你能清晰讲出：为什么需要 client_request_id？
SETNX 和 GET+SET 的区别？
数据库写入失败时，为什么必须删除Redis键？
如何保证已读状态的并发安全？你就已经超越了80%的候选人。技术深度不在于用了多炫的框架，而在于对底层原理的理解和对边缘场景的考量。
还有什么不懂的？评论区留言挨个回 内部邮件系统保姆级教程：面试被问原理？3步搞定核心逻辑
内部邮件系统保姆级教程：面试被问原理？3步搞定核心逻辑
面试被问“内部邮件系统怎么设计”，你答不上来？别慌，这不是背诵题，是考察你对分布式系统、状态机和高并发处理的实战理解。很多人只会调API，一到追问“如何保证消息不丢”、“如何防重放”就卡壳。这篇保姆级教程，不讲虚的，直接带你从零搭建一个能跑、能测、能扛住基本业务量的内部邮件服务。我们不用复杂的大框架，就用Python + FastAPI + Redis + PostgreSQL，把最核心的发送、接收、已读标记、防重逻辑全部拆解清楚。你跟着敲一遍，面试时再遇到类似问题，就能自信地说：“我做过，我知道坑在哪。”
项目目标与核心难点拆解
在动手写代码前，先明确我们要解决什么。内部邮件系统不是简单的“发邮件”，它有几个硬性指标：可靠性（消息必须送达或明确失败）、一致性（用户看到的已读状态必须实时准确）、防重（网络抖动导致的重复请求不能产生两封邮件）。很多初学者容易忽略“防重”和“已读状态的并发更新”，导致线上出现“同一封邮件收到两次”或“明明已读却显示未读”的Bug。
我们的目标很清晰：实现一个最小可用版本（MVP），包含以下四个核心功能：发送接口：支持单发和多发，返回唯一的事务ID。
接收与查询接口：用户只能查自己的收件箱，支持分页。
已读标记接口：用户标记某邮件为已读，需处理并发竞争。
幂等性控制：客户端携带Client-Request-ID，服务端确保同一ID只处理一次。为什么强调幂等性？因为在分布式环境下，前端可能因为超时重试，或者网络层代理转发，导致同一个请求到达后端多次。如果没有幂等控制，用户就会收到重复邮件，这是内部系统的大忌。
目录结构设计
一个清晰的项目结构能让你在面试时快速定位代码，也方便后续扩展。我们采用分层架构，职责分离。
internal-mail-service/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 入口
│ ├── config.py # 配置管理
│ ├── models/ # 数据库模型
│ │ ├── __init__.py
│ │ ├── email.py # Email 表结构
│ │ └── user.py # User 表结构
│ ├── schemas/ # Pydantic 请求/响应模型
│ │ ├── __init__.py
│ │ └── email.py # 数据校验与序列化
│ ├── services/ # 业务逻辑层
│ │ ├── __init__.py
│ │ ├── email_service.py # 核心邮件逻辑
│ │ └── idempotency.py # 幂等性服务
│ ├── api/ # API 路由层
│ │ ├── __init__.py
│ │ └── v1/
│ │ └── emails.py
│ └── utils/ # 工具函数
│ ├── __init__.py
│ └── redis_client.py
├── alembic/ # 数据库迁移
├── tests/ # 单元测试
├── requirements.txt
└── .env关键说明：services/ 层是核心，所有业务逻辑都在这里，API层只做参数校验和调用。
idempotency.py 单独抽离，因为幂等性是一个通用能力，未来其他接口（如转账、下单）也能复用。
使用 alembic 管理数据库迁移，避免直接改表结构导致线上事故。核心代码实现
1. 数据库模型定义
先定义数据结构。邮件表需要记录发送者、接收者、主题、内容、状态（未读/已读/删除）以及时间戳。
# app/models/email.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
from datetime import datetimeclass Email(Base):__tablename__ = emailsid = Column(Integer, primary_key=True, index=True)# 幂等键，用于防重，唯一索引client_request_id = Column(String(128), unique=True, index=True, nullable=False)sender_id = Column(Integer, ForeignKey(users.id), nullable=False)recipient_id = Column(Integer, ForeignKey(users.id), nullable=False, index=True)subject = Column(String(255), nullable=False)content = Column(Text, nullable=False)# 状态：0=未读, 1=已读, 2=删除status = Column(Integer, default=0, index=True)created_at = Column(DateTime, default=datetime.utcnow)read_at = Column(DateTime, nullable=True)# 关联关系，方便查询sender = relationship(User, foreign_keys=[sender_id])recipient = relationship(User, foreign_keys=[recipient_id])注意：client_request_id 加了 unique=True 和 index=True，这是数据库层面的最后一道防重防线。即使Redis失效，数据库也不会插入重复数据。
2. 幂等性服务实现
幂等性的核心思想是：先查，后写，用锁保护。我们使用Redis的 SETNX（Set if Not Exists）命令来实现原子性的“检查并设置”。
# app/services/idempotency.py
import redis
import json
from app.config import settings
from typing import Optionalclass IdempotencyService:def __init__(self):# 初始化 Redis 连接池self.redis_client = redis.Redis(host=settings.REDIS_HOST,port=settings.REDIS_PORT,db=0,decode_responses=True)def check_and_set(self, client_request_id: str, result: dict) - bool:检查是否已处理过该请求。如果未处理，则设置键值并返回True，表示允许继续执行。如果已处理，返回False，并返回之前的结果。key = fidempotency:{client_request_id}# 设置过期时间为24小时，避免Redis数据无限膨胀try:# SETNX: 如果键不存在，则设置并返回1；如果存在，返回0if self.redis_client.setnx(key, json.dumps(result), ex=86400):return Trueelse:# 键已存在，说明重复请求return Falseexcept redis.RedisError as e:# Redis 故障时，降级为数据库唯一约束兜底# 这里可以选择抛异常或记录日志后继续执行print(fRedis error: {e})return Truedef get_previous_result(self, client_request_id: str) - Optional[dict]:获取之前处理的结果key = fidempotency:{client_request_id}data = self.redis_client.get(key)if data:return json.loads(data)return None为什么用 setnx 而不是 get + set？
因为 get 和 set 是两步操作，在高并发下存在竞态条件：两个请求同时 get 发现为空，然后同时 set，导致都以为自己是第一次。setnx 是原子操作，Redis保证只有一个请求能成功设置，其他请求会立即得到失败信号。
3. 邮件发送核心逻辑
这是最复杂的部分，需要结合数据库事务和幂等性。
# app/services/email_service.py
from sqlalchemy.orm import Session
from app.models.email import Email
from app.schemas.email import EmailCreate, EmailResponse
from app.services.idempotency import IdempotencyService
from fastapi import HTTPException, status
from typing import List
import uuidclass EmailService:def __init__(self, db: Session, idempotency_service: IdempotencyService):self.db = dbself.idempotency = idempotency_servicedef send_email(self, sender_id: int, email_in: EmailCreate) - EmailResponse:发送邮件1. 幂等性检查2. 业务校验3. 数据库写入4. 记录幂等结果# 1. 幂等性检查# 如果之前处理过，直接返回之前的结果prev_result = self.idempotency.get_previous_result(email_in.client_request_id)if prev_result:return EmailResponse(**prev_result)# 检查是否允许写入（原子操作）if not self.idempotency.check_and_set(email_in.client_request_id, {}):# 理论上不会走到这里，因为上面已经获取了结果# 但如果 Redis 和 DB 状态不一致，这里兜底raise HTTPException(status_code=409, detail=Duplicate request)try:# 2. 业务校验# 检查接收者是否存在（实际项目中可能支持多收件人，这里简化为单收件人）if not self.db.query(User).filter(User.id == email_in.recipient_id).first():raise HTTPException(status_code=404, detail=Recipient not found)# 3. 创建邮件对象email_obj = Email(client_request_id=email_in.client_request_id,sender_id=sender_id,recipient_id=email_in.recipient_id,subject=email_in.subject,content=email_in.content,status=0 # 默认未读)self.db.add(email_obj)self.db.commit()self.db.refresh(email_obj)# 4. 记录幂等结果，供后续重试返回result = {id: email_obj.id,client_request_id: email_in.client_request_id,status: sent}self.idempotency.set_result(email_in.client_request_id, result)return EmailResponse(id=email_obj.id,subject=email_obj.subject,status=sent)except Exception as e:# 发生异常，回滚事务，并删除幂等键，允许客户端重试self.db.rollback()self.idempotency.delete_key(email_in.client_request_id)raise HTTPException(status_code=500, detail=fFailed to send email: {str(e)})关键细节解析：异常处理：如果数据库写入失败，必须回滚事务，并且删除Redis中的幂等键。否则，客户端重试时会被幂等性拦截，永远无法成功。
结果缓存：成功后，将结果存入Redis。如果客户端因为网络原因没收到响应而重试，服务端直接返回缓存的结果，无需再查数据库，性能极高。4. API路由层
# app/api/v1/emails.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.services.email_service import EmailService
from app.services.idempotency import IdempotencyService
from app.schemas.email import EmailCreate, EmailResponserouter = APIRouter(prefix=/emails, tags=[emails])@router.post(/, response_model=EmailResponse)
def send_email(email_in: EmailCreate,current_user_id: int = Depends(get_current_user), # 假设从Token解析db: Session = Depends(get_db)
):发送邮件接口请求体必须包含 client_request_ididempotency_service = IdempotencyService()email_service = EmailService(db, idempotency_service)return email_service.send_email(current_user_id, email_in)@router.get(/unread, response_model=List[EmailResponse])
def get_unread_emails(skip: int = 0,limit: int = 10,current_user_id: int = Depends(get_current_user),db: Session = Depends(get_db)
):获取未读邮件，支持分页emails = db.query(Email).filter(Email.recipient_id == current_user_id,Email.status == 0).offset(skip).limit(limit).all()return [EmailResponse(id=e.id, subject=e.subject, status=unread) for e in emails]运行与测试
环境准备
安装依赖：
pip install fastapi uvicorn sqlalchemy psycopg2-binary redis pydantic alembic python-dotenv配置 .env 文件：
DATABASE_URL=postgresql://user:pass@localhost:5432/mail_db
REDIS_HOST=localhost
REDIS_PORT=6379启动服务：
uvicorn app.main:app --reload测试幂等性
使用 curl 模拟重复请求：
# 第一次请求
curl -X POST http://localhost:8000/emails/ \-H Content-Type: application/json \-d '{client_request_id: test-123,recipient_id: 1,subject: Hello,content: World}'
# 响应: {id: 1, subject: Hello, status: sent}# 第二次请求，完全相同的 client_request_id
curl -X POST http://localhost:8000/emails/ \-H Content-Type: application/json \-d '{client_request_id: test-123,recipient_id: 1,subject: Hello,content: World}'
# 响应: {id: 1, subject: Hello, status: sent} (ID相同，证明防重成功)测试要点：检查数据库 emails 表，client_request_id 为 test-123 的记录只有一条。
检查Redis，idempotency:test-123 键存在，且值为JSON字符串。
模拟数据库写入失败（如手动停掉PostgreSQL），再发送请求，确认Redis键被删除，恢复数据库后重试能成功。优化扩展与避坑指南
1. 高并发下的已读状态更新
如果多个用户同时标记同一封邮件为已读（虽然内部系统少见，但技术上要防范），直接使用 UPDATE ... SET status=1 会有竞态条件。建议使用乐观锁或Redis位图。
方案A：乐观锁（推荐用于小规模）
在 Email 表中增加 version 字段。
UPDATE emails SET status=1, version=version+1 WHERE id=123 AND version=5;如果影响行数为0，说明被其他请求更新，重试或报错。
方案B：Redis位图（适合海量数据）
用Redis的 SETBIT 命令标记已读。
# key: user:123:read_bits
# bit position: email_id
redis_client.setbit(user:123:read_bits, email_id, 1)查询未读时，结合数据库和Redis位图判断。这种方案性能极高，但需要处理Redis和数据库的一致性。
2. 消息可靠性保障
目前我们的实现是同步写入数据库，如果数据库挂了，请求会失败。对于更高要求，可以引入消息队列（如Kafka、RabbitMQ）：请求先到MQ，返回成功。
消费者从MQ拉取消息，写入数据库。
如果写入失败，消息重新入队，直到成功或进入死信队列。但注意：MQ会引入延迟，且增加系统复杂度。内部系统通常对实时性要求高，同步写入+幂等性已足够。
3. 避免常见坑不要只用Redis防重：Redis可能宕机或数据丢失，必须配合数据库唯一约束。
幂等键要足够唯一：建议使用 UUID + 用户ID + 业务类型 组合，避免跨用户冲突。
超时设置：Redis操作要设置超时，避免网络抖动导致接口卡死。
日志记录：在幂等性检查失败时，记录警告日志，方便排查是否是客户端Bug。小结
通过这个保姆级教程，我们搭建了一个具备生产级基础能力的内部邮件系统。核心不是代码本身，而是幂等性设计和状态一致性处理的思路。面试时，如果你能清晰讲出：为什么需要 client_request_id？
SETNX 和 GET+SET 的区别？
数据库写入失败时，为什么必须删除Redis键？
如何保证已读状态的并发安全？你就已经超越了80%的候选人。技术深度不在于用了多炫的框架，而在于对底层原理的理解和对边缘场景的考量。
还有什么不懂的？评论区留言挨个回