内部邮件系统保姆级教程：面试被问原理？3步搞定核心逻辑

面试被问“内部邮件系统怎么设计”，你答不上来？别慌，这不是背诵题，是考察你对分布式系统、状态机和高并发处理的实战理解。很多人只会调API，一到追问“如何保证消息不丢”、“如何防重放”就卡壳。这篇保姆级教程，不讲虚的，直接带你从零搭建一个能跑、能测、能扛住基本业务量的内部邮件服务。我们不用复杂的大框架，就用Python + FastAPI + Redis + PostgreSQL，把最核心的发送、接收、已读标记、防重逻辑全部拆解清楚。你跟着敲一遍，面试时再遇到类似问题，就能自信地说：“我做过，我知道坑在哪。”

项目目标与核心难点拆解

在动手写代码前，先明确我们要解决什么。内部邮件系统不是简单的“发邮件”，它有几个硬性指标：可靠性（消息必须送达或明确失败）、一致性（用户看到的已读状态必须实时准确）、防重（网络抖动导致的重复请求不能产生两封邮件）。很多初学者容易忽略“防重”和“已读状态的并发更新”，导致线上出现“同一封邮件收到两次”或“明明已读却显示未读”的Bug。

我们的目标很清晰：实现一个最小可用版本（MVP），包含以下四个核心功能：

1. 发送接口：支持单发和多发，返回唯一的事务ID。
2. 接收与查询接口：用户只能查自己的收件箱，支持分页。
3. 已读标记接口：用户标记某邮件为已读，需处理并发竞争。
4. 幂等性控制：客户端携带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__ = "emails"id = 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 = f"idempotency:{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(f"Redis error: {e}")return Truedef get_previous_result(self, client_request_id: str) -> Optional[dict]:"""获取之前处理的结果"""key = f"idempotency:{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=f"Failed 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_id"""idempotency_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相同，证明防重成功)

测试要点：

1. 检查数据库 emails 表，client_request_id 为 test-123 的记录只有一条。
2. 检查Redis，idempotency:test-123 键存在，且值为JSON字符串。
3. 模拟数据库写入失败（如手动停掉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）：

1. 请求先到MQ，返回成功。
2. 消费者从MQ拉取消息，写入数据库。
3. 如果写入失败，消息重新入队，直到成功或进入死信队列。

但注意：MQ会引入延迟，且增加系统复杂度。内部系统通常对实时性要求高，同步写入+幂等性已足够。

3. 避免常见坑

• 不要只用Redis防重：Redis可能宕机或数据丢失，必须配合数据库唯一约束。
• 幂等键要足够唯一：建议使用 UUID + 用户ID + 业务类型 组合，避免跨用户冲突。
• 超时设置：Redis操作要设置超时，避免网络抖动导致接口卡死。
• 日志记录：在幂等性检查失败时，记录警告日志，方便排查是否是客户端Bug。

小结

通过这个保姆级教程，我们搭建了一个具备生产级基础能力的内部邮件系统。核心不是代码本身，而是幂等性设计和状态一致性处理的思路。面试时，如果你能清晰讲出：

1. 为什么需要 client_request_id？
2. SETNX 和 GET+SET 的区别？
3. 数据库写入失败时，为什么必须删除Redis键？
4. 如何保证已读状态的并发安全？

你就已经超越了80%的候选人。技术深度不在于用了多炫的框架，而在于对底层原理的理解和对边缘场景的考量。

还有什么不懂的？评论区留言挨个回