qidao123.com技术社区-IT企服评测·应用市场

标题: 点赞背后的技能大冒险：分布式事务与SAGA模式 [打印本页]
作者: 杀鸡焉用牛刀    时间: 2025-5-7 01:10
标题: 点赞背后的技能大冒险：分布式事务与SAGA模式
title: 点赞背后的技能大冒险：分布式事务与SAGA模式
date: 2025/05/07 00:12:40
updated: 2025/05/07 00:12:40
author: cmdragon
excerpt:
在微服务架构中，点赞操纵涉及多个服务的数据更新，传统数据库事务在分布式系统中失效，需接纳SAGA事务模式。SAGA将事务分解为多个本地事务，通过补偿机制包管最终一致性。每个操纵需定义对应的补偿操纵，补偿操纵需幂等，并记录事务状态和实现超机遇制。代码实现包罗基础模型定义、事务上下文管理器和核心业务逻辑，测试验证正常和异常流程。生产环境中建议添加事务日志、实现定时补偿使命和服务降级策略。
categories:
tags:
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
探索数千个预构建的 AI 应用，开启你的下一个伟大创意：https://tools.cmdragon.cn/
1. 分布式事务的挑衅与办理方案

在微服务架构中，点赞这类看似简单的操纵可能涉及多个服务的数据更新。假设我们有两个微服务：
当用户点赞时，必要同时更新：
传统数据库事务在分布式系统中失效，我们必要接纳SAGA事务模式。这种模式将事务分解为多个本地事务，通过补偿机制包管最终一致性。
2. SAGA事务模式原理

2.1 执行流程示例

正常流程：
  1. [文章服务+1] -> [用户服务创建记录]
复制代码
异常处理：
  1. [文章服务+1] -> [用户服务失败] -> [文章服务-1补偿]
复制代码
2.2 补偿机制要点

3. 实现代码详解

3.1 基础模型定义
  1. # 文章服务模型
  2. class Article(Tortoise.Model):
  3.     id = fields.IntField(pk=True)
  4.     title = fields.CharField(max_length=255)
  5.     likes = fields.IntField(default=0)
  8. # 用户服务模型
  9. class UserLikeRecord(Tortoise.Model):
  10.     id = fields.UUIDField(pk=True)
  11.     user_id = fields.BigIntField()
  12.     article_id = fields.BigIntField()
  13.     created_at = fields.DatetimeField(auto_now_add=True)
  16. # Pydantic响应模型
  17. class LikeResponse(BaseModel):
  18.     article_id: int
  19.     current_likes: int
  20.     user_record_id: UUID
复制代码
3.2 事务上下文管理器
  1. class SagaTransaction:
  2.     def __init__(self):
  3.         self.compensation_actions = []
  5.     async def __aenter__(self):
  6.         return self
  8.     async def __aexit__(self, exc_type, exc, traceback):
  9.         if exc_type is not None:
  10.             await self.compensate()
  12.     def add_compensation(self, coro_func, *args):
  13.         self.compensation_actions.append((coro_func, args))
  15.     async def compensate(self):
  16.         for coro_func, args in reversed(self.compensation_actions):
  17.             try:
  18.                 await coro_func(*args)
  19.             except Exception as e:
  20.                 logging.error(f"Compensation failed: {str(e)}")
复制代码
3.3 核心业务实现
  1. @app.post("/articles/{article_id}/like", response_model=LikeResponse)
  2. async def like_article(
  3.         article_id: int,
  4.         user_id: int = Header(..., alias="X-User-ID")
  5. ):
  6.     async with SagaTransaction() as saga:
  7.         # 第一步：更新文章点赞数
  8.         article = await Article.get(id=article_id)
  9.         original_likes = article.likes
  10.         article.likes += 1
  11.         await article.save()
  13.         # 记录补偿操作（回滚点赞数）
  14.         saga.add_compensation(
  15.             self.compensate_article_likes,
  16.             article_id,
  17.             original_likes
  18.         )
  20.         # 第二步：创建用户点赞记录
  21.         try:
  22.             record = await UserLikeRecord.create(
  23.                 user_id=user_id,
  24.                 article_id=article_id
  25.             )
  26.         except Exception as e:
  27.             # 自动触发补偿流程
  28.             raise HTTPException(500, "Like record creation failed")
  30.         # 记录补偿操作（删除记录）
  31.         saga.add_compensation(
  32.             self.compensate_user_record,
  33.             record.id
  34.         )
  36.         return LikeResponse(
  37.             article_id=article_id,
  38.             current_likes=article.likes,
  39.             user_record_id=record.id
  40.         )
  43. # 补偿方法示例
  44. async def compensate_article_likes(article_id: int, original_count: int):
  45.     article = await Article.get(id=article_id)
  46.     article.likes = original_count
  47.     await article.save()
  50. async def compensate_user_record(record_id: UUID):
  51.     await UserLikeRecord.filter(id=record_id).delete()
复制代码
4. 测试与验证

4.1 正常流程测试
  1. async def test_successful_like():
  2.     async with AsyncClient(app=app, base_url="http://test") as ac:
  3.         response = await ac.post(
  4.             "/articles/1/like",
  5.             headers={"X-User-ID": "123"}
  6.         )
  7.         assert response.status_code == 200
  8.         data = response.json()
  9.         assert data["current_likes"] == 1
复制代码
4.2 异常流程测试
  1. async def test_failed_transaction():
  2.     with patch("UserLikeRecord.create", side_effect=Exception("DB Error")):
  3.         response = await ac.post(
  4.             "/articles/1/like",
  5.             headers={"X-User-ID": "123"}
  6.         )
  7.         assert response.status_code == 500
  9.         # 验证补偿是否执行
  10.         article = await Article.get(id=1)
  11.         assert article.likes == 0
复制代码
5. 课后Quiz

Q1：为什么补偿操纵必要设计为幂等？
A. 提高系统性能
B. 防止重复补偿导致数据错误
C. 减少数据库连接数
D. 满意HTTP协议规范
正确答案：B
分析：网络重试可能导致补偿操纵被多次触发，幂等设计确保多次执行效果一致，制止数据不一致。
Q2：以下哪些情况必要触发补偿机制？（多选）
A. 用户服务数据库连接超时
B. 文章不存在返回404错误
C. 用户重复点赞
D. 数据库主从同步延迟
正确答案：A
分析：404属于业务校验错误应在事务开始前检查，重复点赞属于业务逻辑错误，主从同步属于基础架构题目。只有跨服务操纵失败必要补偿。
6. 常见报错与办理方案

报错1：TransactionManagementError - 事务已关闭
  1. TransactionManagementError: Transaction already closed
复制代码
原因：异步上下文管理器中过早关闭数据库连接
办理方案：
报错2：HTTP 422 Unprocessable Entity
  1. {
  2.   "detail": "Field required"
  3. }
复制代码
原因：请求体缺少必要字段或类型不匹配
办理方案：
报错3：TimeoutError - 数据库操纵超时
  1. TimeoutError: Connection pool exhausted
复制代码
原因：数据库连接池不足或查询未优化
办理方案：
  1. TORTOISE_CONFIG["connections"]["default"]["pool_size"] = 20
复制代码
7. 生产环境建议

  1. class TransactionLog(Tortoise.Model):
  2.     transaction_id = fields.UUIDField()
  3.     service_name = fields.CharField(max_length=50)
  4.     action_type = fields.CharField(max_length=20)  # main/compensation
  5.     status = fields.CharField(max_length=10)  # pending/done/failed
  6.     created_at = fields.DatetimeField(auto_now_add=True)
复制代码
  1. async def check_hanging_transactions():
  2.     # 查找超过5分钟未完成的事务
  3.     pending = await TransactionLog.filter(
  4.         status="pending",
  5.         created_at__lt=datetime.now() - timedelta(minutes=5)
  6.     )
  8.     for transaction in pending:
  9.         # 执行补偿逻辑
  10.         await retry_compensation(transaction)
复制代码

（完备示例代码需配合PostgreSQL数据库运行，安装依赖：fastapi uvicorn tortoise-orm httpx python-multipart）
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长，阅读完备的文章：点赞背后的技能大冒险：分布式事务与SAGA模式 | cmdragon's Blog
往期文章归档：


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/) Powered by Discuz! X3.4