点赞背后的技术大冒险：分布式事务与SAGA模式

技术分享 9个月前 (05-07) 0 999+

title: 点赞背后的技术大冒险：分布式事务与SAGA模式
date: 2025/05/07 00:12:40
updated: 2025/05/07 00:12:40
author: cmdragon

excerpt:
在微服务架构中，点赞操作涉及多个服务的数据更新，传统数据库事务在分布式系统中失效，需采用SAGA事务模式。SAGA将事务分解为多个本地事务，通过补偿机制保证最终一致性。每个操作需定义对应的补偿操作，补偿操作需幂等，并记录事务状态和实现超时机制。代码实现包括基础模型定义、事务上下文管理器和核心业务逻辑，测试验证正常和异常流程。生产环境中建议添加事务日志、实现定时补偿任务和服务降级策略。

categories:

后端开发
FastAPI

tags:

分布式事务
SAGA模式
微服务架构
补偿机制
Python实现
事务管理
数据库操作

点赞背后的技术大冒险：分布式事务与SAGA模式

扫描二维码
关注或者微信搜一搜：编程智域前端至全栈交流与成长

探索数千个预构建的 AI 应用，开启你的下一个伟大创意：https://tools.cmdragon.cn/

1. 分布式事务的挑战与解决方案

在微服务架构中，点赞这类看似简单的操作可能涉及多个服务的数据更新。假设我们有两个微服务：

文章服务（存储文章信息和点赞数）
用户服务（记录用户点赞行为）

当用户点赞时，需要同时更新：

文章服务的点赞计数器
用户服务的点赞记录

传统数据库事务在分布式系统中失效，我们需要采用SAGA事务模式。这种模式将事务分解为多个本地事务，通过补偿机制保证最终一致性。

2. SAGA事务模式原理

2.1 执行流程示例

正常流程：

[文章服务+1] -> [用户服务创建记录]

异常处理：

[文章服务+1] -> [用户服务失败] -> [文章服务-1补偿]

2.2 补偿机制要点

每个操作必须定义对应的补偿操作
补偿操作需要幂等（重复执行结果一致）
必须记录事务状态
需要实现事务超时机制

3. 实现代码详解

3.1 基础模型定义

# 文章服务模型 class Article(Tortoise.Model):     id = fields.IntField(pk=True)     title = fields.CharField(max_length=255)     likes = fields.IntField(default=0)   # 用户服务模型 class UserLikeRecord(Tortoise.Model):     id = fields.UUIDField(pk=True)     user_id = fields.BigIntField()     article_id = fields.BigIntField()     created_at = fields.DatetimeField(auto_now_add=True)   # Pydantic响应模型 class LikeResponse(BaseModel):     article_id: int     current_likes: int     user_record_id: UUID

3.2 事务上下文管理器

class SagaTransaction:     def __init__(self):         self.compensation_actions = []      async def __aenter__(self):         return self      async def __aexit__(self, exc_type, exc, traceback):         if exc_type is not None:             await self.compensate()      def add_compensation(self, coro_func, *args):         self.compensation_actions.append((coro_func, args))      async def compensate(self):         for coro_func, args in reversed(self.compensation_actions):             try:                 await coro_func(*args)             except Exception as e:                 logging.error(f"Compensation failed: {str(e)}")

3.3 核心业务实现

@app.post("/articles/{article_id}/like", response_model=LikeResponse) async def like_article(         article_id: int,         user_id: int = Header(..., alias="X-User-ID") ):     async with SagaTransaction() as saga:         # 第一步：更新文章点赞数         article = await Article.get(id=article_id)         original_likes = article.likes         article.likes += 1         await article.save()          # 记录补偿操作（回滚点赞数）         saga.add_compensation(             self.compensate_article_likes,             article_id,             original_likes         )          # 第二步：创建用户点赞记录         try:             record = await UserLikeRecord.create(                 user_id=user_id,                 article_id=article_id             )         except Exception as e:             # 自动触发补偿流程             raise HTTPException(500, "Like record creation failed")          # 记录补偿操作（删除记录）         saga.add_compensation(             self.compensate_user_record,             record.id         )          return LikeResponse(             article_id=article_id,             current_likes=article.likes,             user_record_id=record.id         )   # 补偿方法示例 async def compensate_article_likes(article_id: int, original_count: int):     article = await Article.get(id=article_id)     article.likes = original_count     await article.save()   async def compensate_user_record(record_id: UUID):     await UserLikeRecord.filter(id=record_id).delete()

4. 测试与验证

4.1 正常流程测试

async def test_successful_like():     async with AsyncClient(app=app, base_url="http://test") as ac:         response = await ac.post(             "/articles/1/like",             headers={"X-User-ID": "123"}         )         assert response.status_code == 200         data = response.json()         assert data["current_likes"] == 1

4.2 异常流程测试

async def test_failed_transaction():     with patch("UserLikeRecord.create", side_effect=Exception("DB Error")):         response = await ac.post(             "/articles/1/like",             headers={"X-User-ID": "123"}         )         assert response.status_code == 500          # 验证补偿是否执行         article = await Article.get(id=1)         assert article.likes == 0

5. 课后Quiz

Q1：为什么补偿操作需要设计为幂等？
A. 提高系统性能
B. 防止重复补偿导致数据错误
C. 减少数据库连接数
D. 满足HTTP协议规范

正确答案：B
解析：网络重试可能导致补偿操作被多次触发，幂等设计确保多次执行结果一致，避免数据不一致。

Q2：以下哪些情况需要触发补偿机制？（多选）
A. 用户服务数据库连接超时
B. 文章不存在返回404错误
C. 用户重复点赞
D. 数据库主从同步延迟

正确答案：A
解析：404属于业务校验错误应在事务开始前检查，重复点赞属于业务逻辑错误，主从同步属于基础架构问题。只有跨服务操作失败需要补偿。

6. 常见报错与解决方案

报错1：TransactionManagementError - 事务已关闭

TransactionManagementError: Transaction already closed

原因：异步上下文管理器中过早关闭数据库连接
解决方案：

检查事务作用域范围
确保所有数据库操作在同一个事务上下文中
更新Tortoise-ORM到最新版本

报错2：HTTP 422 Unprocessable Entity

{   "detail": "Field required" }

原因：请求体缺少必要字段或类型不匹配
解决方案：

检查请求头是否包含X-User-ID
验证URL参数类型是否正确
使用Swagger文档测试接口格式

报错3：TimeoutError - 数据库操作超时

TimeoutError: Connection pool exhausted

原因：数据库连接池不足或查询未优化
解决方案：

增加连接池大小配置：

TORTOISE_CONFIG["connections"]["default"]["pool_size"] = 20

为高频查询字段添加索引
使用select_related优化关联查询

7. 生产环境建议

添加事务日志：

class TransactionLog(Tortoise.Model):     transaction_id = fields.UUIDField()     service_name = fields.CharField(max_length=50)     action_type = fields.CharField(max_length=20)  # main/compensation     status = fields.CharField(max_length=10)  # pending/done/failed     created_at = fields.DatetimeField(auto_now_add=True)

实现定时补偿任务：

async def check_hanging_transactions():     # 查找超过5分钟未完成的事务     pending = await TransactionLog.filter(         status="pending",         created_at__lt=datetime.now() - timedelta(minutes=5)     )      for transaction in pending:         # 执行补偿逻辑         await retry_compensation(transaction)