基于本地消息表实现分布式事务

假设我们有一个电商系统,包含订单服务和库存服务。当用户下单时,需要在订单服务中创建订单,同时在库存服务中扣减库存。这是一个典型的分布式事务场景,我们需要保证这两个操作要么都成功,要么都失败,以保证数据的最终一致性。

项目结构:

  1. 订单服务(Order Service)
  2. 库存服务(Inventory Service)
  3. 本地消息表(Local Message Table)
  4. 消息恢复系统(Message Recovery System)

核心思想:
使用本地消息表来实现分布式事务。在订单服务中,我们将创建订单和发送消息这两个操作放在一个本地事务中。如果本地事务成功,则订单创建成功,消息也被保存到本地消息表中。然后通过定时任务或消息队列来发送消息到库存服务,实现库存扣减。如果在这个过程中出现任何异常,我们可以通过重试机制来保证最终一致性。

下面是详细的代码实现:

订单服务(Order Service)

@Service  
@Transactional  
public class OrderService {  @Autowired  private OrderRepository orderRepository;  @Autowired  private LocalMessageRepository localMessageRepository;  @Autowired  private KafkaTemplate<String, String> kafkaTemplate;  public void createOrder(Order order) {  // 开启本地事务  TransactionStatus txStatus = transactionManager.getTransaction(new DefaultTransactionDefinition());  try {  // 1. 保存订单  orderRepository.save(order);  // 2. 创建本地消息  LocalMessage message = new LocalMessage();  message.setMessageId(UUID.randomUUID().toString());  message.setMessage(JSON.toJSONString(order));  message.setStatus("NEW");  localMessageRepository.save(message);  // 3. 提交事务  transactionManager.commit(txStatus);  // 4. 发送消息到Kafka  kafkaTemplate.send("inventory-topic", message.getMessageId(), message.getMessage());  } catch (Exception e) {  // 回滚事务  transactionManager.rollback(txStatus);  throw new RuntimeException("Create order failed", e);  }  }  
}

库存服务(Inventory Service)

@Service  
public class InventoryService {  @Autowired  private InventoryRepository inventoryRepository;  @KafkaListener(topics = "inventory-topic")  public void handleOrderCreation(ConsumerRecord<String, String> record) {  String messageId = record.key();  Order order = JSON.parseObject(record.value(), Order.class);  try {  // 扣减库存  inventoryRepository.decreaseStock(order.getProductId(), order.getQuantity());  // 确认消息处理成功  kafkaTemplate.send("inventory-result-topic", messageId, "SUCCESS");  } catch (Exception e) {  // 消息处理失败,发送失败消息  kafkaTemplate.send("inventory-result-topic", messageId, "FAILED");  }  }  
}

本地消息表(Local Message Table)

@Entity  
@Table(name = "local_message")  
public class LocalMessage {  @Id  private String messageId;  private String message;  private String status; // NEW, SENT, CONFIRMED  private Date createTime;  private Date updateTime;  // Getters and setters  
}

消息恢复系统(Message Recovery System)

@Component  
public class MessageRecoverySystem {  @Autowired  private LocalMessageRepository localMessageRepository;  @Autowired  private KafkaTemplate<String, String> kafkaTemplate;  @Scheduled(fixedRate = 60000) // 每分钟执行一次  public void recoverFailedMessages() {  List<LocalMessage> failedMessages = localMessageRepository.findByStatusAndCreateTimeBefore("NEW", new Date(System.currentTimeMillis() - 300000)); // 5分钟前的消息  for (LocalMessage message : failedMessages) {  try {  kafkaTemplate.send("inventory-topic", message.getMessageId(), message.getMessage());  message.setStatus("SENT");  localMessageRepository.save(message);  } catch (Exception e) {  // 记录日志,等待下次重试  log.error("Failed to recover message: " + message.getMessageId(), e);  }  }  }  @KafkaListener(topics = "inventory-result-topic")  public void handleInventoryResult(ConsumerRecord<String, String> record) {  String messageId = record.key();  String result = record.value();  LocalMessage message = localMessageRepository.findById(messageId).orElse(null);  if (message != null) {  if ("SUCCESS".equals(result)) {  message.setStatus("CONFIRMED");  } else {  message.setStatus("FAILED");  }  localMessageRepository.save(message);  }  }  
}

代码说明:

  1. 订单服务:
    • 在一个本地事务中完成订单创建和本地消息保存。
    • 事务成功后,立即发送消息到Kafka。
  2. 库存服务:
    • 监听Kafka消息,处理库存扣减。
    • 处理结果(成功或失败)通过Kafka反馈给订单服务。
  3. 本地消息表:
    • 存储待发送的消息,包括消息ID、内容、状态等信息。
  4. 消息恢复系统:
    • 定期检查本地消息表,重新发送失败的消息。
    • 监听库存服务的处理结果,更新本地消息状态。

项目亮点:

  1. 高可用性: 即使在网络故障或服务宕机的情况下,也能保证消息最终被成功处理。
  2. 数据一致性: 通过本地事务保证订单创建和消息发送的原子性,再通过消息重试机制保证最终一致性。
  3. 解耦性: 订单服务和库存服务通过消息进行异步通信,降低了系统耦合度。
  4. 可靠性: 使用本地消息表作为消息队列的可靠存储,避免了消息丢失的风险。
  5. 扩展性: 该方案易于扩展,可以方便地增加新的微服务而不影响现有服务。
  6. 性能: 采用异步处理方式,提高了系统的整体吞吐量。

通过这种方式,我们实现了在分布式系统中保证数据最终一致性的目标,同时保持了系统的高可用性和可扩展性。这种方案特别适用于对实时性要求不是特别高,但对数据一致性有较高要求的业务场景。


系列文章

  1. IT Governance Framework:IT治理框架
  2. 12306亿级流量架构分析(史上最全)
  3. 京东内部Redis性能优化最佳实践
  4. 金融级多数据中心灾备互联
  5. TOGAF业务架构-CSDN博客
  6. 如何建设金融数据中心-CSDN博客

互联网Java架构师-CSDN博客


资料下载和预览地址:

  • 链接: https://pan.baidu.com/s/1LFyFlsIHCv46DBQRfMGP9A 提取码: kx6b 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/893265.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

『 实战项目 』Cloud Backup System - 云备份

文章目录 云备份项目服务端功能服务端功能模块划分客户端功能客户端模块划分 项目条件Jsoncpp第三方库Bundle第三方库httplib第三方库Request类Response类Server类Client类搭建简单服务器搭建简单客户端 服务端工具类实现 - 文件实用工具类服务器配置信息模块实现- 系统配置信息…

网络编程 | UDP组播通信

1、什么是组播 在上一篇博客中&#xff0c;对UDP的广播通信进行了由浅入深的总结梳理&#xff0c;本文继续对UDP的知识体系进行探讨&#xff0c;旨在将UDP的组播通信由浅入深的讲解清楚。 组播是介于单播与广播之间&#xff0c;在一个局域网内&#xff0c;将某些主机添加到组中…

第9章:Python TDD解决货币对象相等性比较难题

写在前面 这本书是我们老板推荐过的&#xff0c;我在《价值心法》的推荐书单里也看到了它。用了一段时间 Cursor 软件后&#xff0c;我突然思考&#xff0c;对于测试开发工程师来说&#xff0c;什么才更有价值呢&#xff1f;如何让 AI 工具更好地辅助自己写代码&#xff0c;或许…

【无标题】微调是迁移学习吗?

是的&#xff0c;微调&#xff08;Fine-Tuning&#xff09;可以被视为一种迁移学习&#xff08;Transfer Learning&#xff09;的形式。迁移学习是一种机器学习方法&#xff0c;其核心思想是利用在一个任务上学到的知识来改进另一个相关任务的性能。微调正是通过在预训练模型的…

Jenkins-获取build用户信息

需求&#xff1a; 代码发布后&#xff0c;将发布结果发送至相关运维同学邮箱&#xff0c;需要获取发布人的信息。jenkins默认是没有相关内置变量的。 需要通过插件的方式进行解决&#xff1a; 插件&#xff1a; user build vars plugin 部署后&#xff0c;可使用的变量&…

【HarmonyOS NAPI 深度探索12】创建你的第一个 HarmonyOS NAPI 模块

【HarmonyOS NAPI 深度探索12】创建你的第一个 HarmonyOS NAPI 模块 在本篇文章中&#xff0c;我们将一步步走过如何创建一个简单的 HarmonyOS NAPI 模块。通过这个模块&#xff0c;你将能够更好地理解 NAPI 的工作原理&#xff0c;并在你的应用中开始使用 C 与 JavaScript 的…

OpenCV相机标定与3D重建(62)根据两个投影矩阵和对应的图像点来计算3D空间中点的坐标函数triangulatePoints()的使用

加粗样式- 操作系统&#xff1a;ubuntu22.04 OpenCV版本&#xff1a;OpenCV4.9 IDE:Visual Studio Code 编程语言&#xff1a;C11 算法描述 这个函数通过使用立体相机对3维点的观测&#xff0c;重建这些点的三维坐标&#xff08;以齐次坐标表示&#xff09;。 cv::triangula…

为AI聊天工具添加一个知识系统 之58 从文字块构造到数据库创建

本文要点 要点 文字块 为了项目(为AI聊天工具的聊天者 开挂知识系统) ,我们探索一下知识的“种子”到底“藏”在哪&#xff0c;知识树又是如何“生根发芽开会结果”的。 看看下面的四组表达&#xff08;仔细体会理解消化每一个字 以及上下文 和标题&#xff09;&#xff1a;…

【电视盒子】HI3798MV300刷机教程笔记/备份遥控码修复遥控器/ADB/线刷卡刷/电视盒子安装第三方应用软件

心血来潮&#xff0c;看到电视机顶盒满天飞的广告&#xff0c;想改造一下家里的电视盒子&#xff0c;学一下网上的人刷机&#xff0c;但是一切都不知道怎么开始&#xff0c;虽然折腾了一天&#xff0c;以失败告终&#xff0c;还是做点刷机笔记。 0.我的机器 年少不会甄别&…

Python基于OpenCV和PyQt5的人脸识别上课签到系统【附源码】

博主介绍&#xff1a;✌Java老徐、7年大厂程序员经历。全网粉丝12w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专栏推荐订阅&#x1f447;&…

【FPGA】MIPS 12条整数指令【1】

目录 修改后的仿真结果 修改后的完整代码 实现bgtz、bltz、jalr 仿真结果&#xff08;有问题&#xff09; bltz------并未跳转&#xff0c;jCe&#xff1f; 原因是该条跳转语句判断的寄存器r7&#xff0c;在该时刻并未被赋值 代码&#xff08;InstMem修改前&#xff09; i…

uniapp(小程序、app、微信公众号、H5)预览下载文件(pdf)

1. 小程序、app 在uniapp开发小程序环境或者app环境中,都可以使用以下方式预览文件 之前其实写过一篇,就是使用uniapp官网提供文件下载、文件保存、文件打开的API, uniapp文件下载 感兴趣也可以去看下 uni.downloadFile({// baseURL 是

Java面试专题——常见面试题1

引入 本文属于专题中的常见面试题模块&#xff0c;属于面试时经常遇到的&#xff0c;适合需要面试的小伙伴做面试前复习准备用&#xff0c;后续会持续补充 1.面向对象基本特征 面向对象的基本特征是什么&#xff1f;怎么理解&#xff1f; 面向对象的基本特征是封装、继承、…

【博客之星2024】技术洞察:前沿技术趋势与创新实践

前言 随着科技的飞速发展&#xff0c;技术的前沿趋势和跨领域的融合创新已经成为影响现代社会的重要因素。无论是 人工智能 (AI)、区块链 还是 量子计算&#xff0c;这些新兴技术不仅仅改变了行业的运行方式&#xff0c;还深入影响了社会、文化及个人的生活方式。因此&#xf…

VUE实现简单留言板(Timeline+infinite scroll+Springboot+Hibernate)

先贴出效果图&#xff1a; 留言按照倒序排列。在底部的文本框内输入留言后&#xff0c;点击“留言”按钮&#xff0c;留言将保存至数据库中&#xff0c;同时刷新网页&#xff0c;新留言出现在顶部。 当滚动条到底部时&#xff0c;自动调用加载函数&#xff0c;显示更多早期留…

Java基础(3)

Java 数据类型详解 九、运算符 1. 基本运算符 Java 提供了多种运算符来执行不同的操作&#xff1a; 算术运算符&#xff1a;&#xff08;加&#xff09;、-&#xff08;减&#xff09;、*&#xff08;乘&#xff09;、/&#xff08;除&#xff09;、%&#xff08;取模&…

电力场景红外测温图像绝缘套管分割数据集labelme格式2436张1类别

数据集格式&#xff1a;labelme格式(不包含mask文件&#xff0c;仅仅包含jpg图片和对应的json文件) 图片数量(jpg文件个数)&#xff1a;2436 标注数量(json文件个数)&#xff1a;2436 标注类别数&#xff1a;1 标注类别名称:["arrester"] 每个类别标注的框数&am…

降维算法:主成分分析

主成分分析 一种常用的数据分析技术&#xff0c;主要用于数据降维&#xff0c;在众多领域如统计学、机器学习、信号处理等都有广泛应用。 主成分分析是一种通过正交变换将一组可能存在相关性的变量转换为一组线性不相关的变量&#xff08;即主成分&#xff09;的方法。这些主…

CTE与临时表:优劣势对比及使用场景分析

在数据库开发中&#xff0c;尤其是在复杂查询和优化中&#xff0c;**公共表表达式&#xff08;CTE&#xff09;和临时表&#xff08;Temporary Table&#xff09;**是两种常用的工具。尽管它们的功能有些相似&#xff0c;都是为了处理中间结果集&#xff0c;但它们的优劣势和使…

文献综述相关ChatGPT提示词分享

文献综述 ChatGPT 可以帮助提高文献综述的有效性和全面性。ChatGPT可以高效搜索和审查与宝子们课题研究相关的文献资料来源。一些给力的插件工具还可以帮助您总结复杂的研究论文并提取信息以更快更好地消化信息。合理的运用ChatGPT和GPTs可以提高文献综述的清晰度和质量&#…