1. 树叶云首页
  2. 技术资讯

SpringBoot分布式事务之最大努力通知

运维••技术资讯

SpringBoot分布式事务之最大努力通知

作者：FastCoder 2021-04-16 16:02:13

开发

前端

分布式 最大努力通知方案的目标 ：发起通知方通过一定的机制最大努力将业务处理结果通知到接收方。

[[393657]]

环境：springboot.2.4.9 + RabbitMQ3.7.4

什么是最大努力通知

这是一个充值的案例

交互流程 ：

1、账户系统调用充值系统接口。

2、充值系统完成支付向账户系统发起充值结果通知 若通知失败，则充值系统按策略进行重复通知。

3、账户系统接收到充值结果通知修改充值状态。

4、账户系统未接收到通知会主动调用充值系统的接口查询充值结果。

通过上边的例子我们总结最大努力通知方案的目标 ： 发起通知方通过一定的机制最大努力将业务处理结果通知到接收方。 具体包括 ：

1、有一定的消息重复通知机制。 因为接收通知方可能没有接收到通知，此时要有一定的机制对消息重复通知。

2、消息校对机制。 如果尽最大努力也没有通知到接收方，或者接收方消费消息后要再次消费，此时可由接收方主动向通知方查询消息信息来满足需求。

最大努力通知与可靠消息一致性有什么不同?

1、解决方案思想不同 可靠消息一致性，发起通知方需要保证将消息发出去，并且将消息发到接收通知方，消息的可靠性关键由发起通知方来保证。 最大努力通知，发起通知方尽最大的努力将业务处理结果通知为接收通知方，但是可能消息接收不到，此时需要接收通知方主动调用发起通知方的接口查询业务处理结果，通知的可靠性关键在接收通知方。

2、两者的业务应用场景不同 可靠消息一致性关注的是交易过程的事务一致，以异步的方式完成交易。 最大努力通知关注的是交易后的通知事务，即将交易结果可靠的通知出去。

3、技术解决方向不同 可靠消息一致性要解决消息从发出到接收的一致性，即消息发出并且被接收到。 最大努力通知无法保证消息从发出到接收的一致性，只提供消息接收的可靠性机制。可靠机制是，最大努力地将消息通知给接收方，当消息无法被接收方接收时，由接收方主动查询消费。

通过RabbitMQ实现最大努力通知

关于RabbitMQ相关文章《SpringBoot RabbitMQ消息可靠发送与接收 》，《RabbitMQ消息确认机制confirm 》。

 

项目结构

两个子模块users-mananger(账户模块)，pay-manager(支付模块)

依赖

  1. <dependency> 
  2.  <groupId>org.springframework.boot</groupId> 
  3.  <artifactId>spring-boot-starter-data-jpa</artifactId> 
  4. </dependency> 
  5. <dependency> 
  6.  <groupId>org.springframework.boot</groupId> 
  7.  <artifactId>spring-boot-starter-web</artifactId> 
  8. </dependency> 
  9. <dependency> 
  10.  <groupId>org.springframework.boot</groupId> 
  11.  <artifactId>spring-boot-starter-amqp</artifactId> 
  12. </dependency> 
  13. <dependency> 
  14.  <groupId>mysql</groupId> 
  15.  <artifactId>mysql-connector-java</artifactId> 
  16.  <scope>runtime</scope> 
  17. </dependency> 

子模块pay-manager

配置文件

  1. server: 
  2.   port: 8080 
  3. --- 
  4. spring: 
  5.   rabbitmq: 
  6.     host: localhost 
  7.     port: 5672 
  8.     username: guest 
  9.     password: guest 
  10.     virtual-host: / 
  11.     publisherConfirmType: correlated 
  12.     publisherReturns: true 
  13.     listener: 
  14.       simple: 
  15.         concurrency: 5 
  16.         maxConcurrency: 10 
  17.         prefetch: 5 
  18.         acknowledgeMode: MANUAL 
  19.         retry: 
  20.           enabled: true 
  21.           initialInterval: 3000 
  22.           maxAttempts: 3 
  23.         defaultRequeueRejected: false 

实体类

记录充值金额及账户信息

  1. @Entity 
  2. @Table(name = "t_pay_info") 
  3. public class PayInfo implements Serializable{ 
  4.  @Id 
  5.  private Long id; 
  6.  private BigDecimal money ; 
  7.  private Long accountId ; 
  8. }   

DAO及Service

  1. public interface PayInfoRepository extends JpaRepository<PayInfo, Long> { 
  2.  PayInfo findByOrderId(String orderId) ; 
  3. } 
  1. @Service 
  2. public class PayInfoService { 
  3.      
  4.     @Resource 
  5.     private PayInfoRepository payInfoRepository ; 
  6.     @Resource 
  7.     private RabbitTemplate rabbitTemplate ; 
  8.      
  9.   // 数据保存完后发送消息（这里发送消息可以应用确认模式或事物模式） 
  10.     @Transactional 
  11.     public PayInfo savePayInfo(PayInfo payInfo) { 
  12.         payInfo.setId(System.currentTimeMillis()) ; 
  13.         PayInfo result = payInfoRepository.save(payInfo) ; 
  14.         CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString().replaceAll("-", "")) ; 
  15.         try { 
  16.             rabbitTemplate.convertAndSend("pay-exchange", "pay.#", new ObjectMapper().writeValueAsString(payInfo), correlationData) ; 
  17.         } catch (AmqpException | JsonProcessingException e) { 
  18.             e.printStackTrace(); 
  19.         } 
  20.         return result ; 
  21.     } 
  22.      
  23.     public PayInfo queryByOrderId(String orderId) { 
  24.         return payInfoRepository.findByOrderId(orderId) ; 
  25.     } 
  26.      
  27. } 

支付完成后发送消息。

Controller接口

  1. @RestController 
  2. @RequestMapping("/payInfos") 
  3. public class PayInfoController { 
  4.  @Resource 
  5.  private PayInfoService payInfoService ; 
  6.      
  7.   // 支付接口 
  8.  @PostMapping("/pay") 
  9.  public Object pay(@RequestBody PayInfo payInfo) { 
  10.   payInfoService.savePayInfo(payInfo) ; 
  11.   return "支付已提交，等待结果" ; 
  12.  } 
  13.      
  14.  @GetMapping("/queryPay") 
  15.  public Object queryPay(String orderId) { 
  16.   return payInfoService.queryByOrderId(orderId) ; 
  17.  } 
  18.      
  19. } 

子模块users-manager

应用配置

  1. server: 
  2.   port: 8081 
  3. --- 
  4. spring: 
  5.   rabbitmq: 
  6.     host: localhost 
  7.     port: 5672 
  8.     username: guest 
  9.     password: guest 
  10.     virtual-host: / 
  11.     publisherConfirmType: correlated 
  12.     publisherReturns: true 
  13.     listener: 
  14.       simple: 
  15.         concurrency: 5 
  16.         maxConcurrency: 10 
  17.         prefetch: 5 
  18.         acknowledgeMode: MANUAL 
  19.         retry: 
  20.           enabled: true 
  21.           initialInterval: 3000 
  22.           maxAttempts: 3 
  23.         defaultRequeueRejected: false 

实体类

  1. @Entity 
  2. @Table(name = "t_users") 
  3. public class Users { 
  4.  @Id 
  5.  private Long id; 
  6.  private String name ; 
  7.  private BigDecimal money ; 
  8. }   

账户信息表

  1. @Entity 
  2. @Table(name = "t_users_log") 
  3. public class UsersLog { 
  4.  @Id 
  5.  private Long id; 
  6.  private String orderId ; 
  7.  // 0： 支付中，1：已支付，2：已取消 
  8.  @Column(columnDefinition = "int default 0") 
  9.  private Integer status = 0 ; 
  10.  private BigDecimal money ; 
  11.  private Date createTime ; 
  12. } 

账户充值记录表(去重)

DAO及Service

  1. public interface UsersRepository extends JpaRepository<Users, Long> { 
  2. } 
  3. public interface UsersLogRepository extends JpaRepository<UsersLog, Long> { 
  4.  UsersLog findByOrderId(String orderId) ; 
  5. } 

Service类

  1. @Service 
  2. public class UsersService {  
  3.     @Resource 
  4.     private UsersRepository usersRepository ; 
  5.     @Resource 
  6.     private UsersLogRepository usersLogRepository ; 
  7.      
  8.  @Transactional 
  9.  public boolean updateMoneyAndLogStatus(Long id, String orderId) { 
  10.   UsersLog usersLog = usersLogRepository.findByOrderId(orderId) ; 
  11.   if (usersLog != null && 1 == usersLog.getStatus()) { 
  12.    throw new RuntimeException("已支付") ; 
  13.   } 
  14.   Users users = usersRepository.findById(id).orElse(null) ; 
  15.   if (users == null) { 
  16.    throw new RuntimeException("账户不存在") ; 
  17.   } 
  18.   users.setMoney(users.getMoney().add(usersLog.getMoney())) ; 
  19.   usersRepository.save(users) ; 
  20.   usersLog.setStatus(1) ; 
  21.   usersLogRepository.save(usersLog) ; 
  22.   return true ; 
  23.  } 
  24.      
  25.  @Transactional 
  26.  public boolean saveLog(UsersLog usersLog) { 
  27.   usersLog.setId(System.currentTimeMillis()) ; 
  28.   usersLogRepository.save(usersLog) ; 
  29.   return true ; 
  30.  } 
  31. } 

消息监听

  1. @Component 
  2. public class PayMessageListener { 
  3.      
  4.  private static final Logger logger = LoggerFactory.getLogger(PayMessageListener.class) ; 
  5.      
  6.  @Resource 
  7.  private  UsersService usersService ; 
  8.      
  9.  @SuppressWarnings("unchecked") 
  10.  @RabbitListener(queues = {"pay-queue"}) 
  11.  @RabbitHandler 
  12.  public void receive(Message message, Channel channel) { 
  13.   long deliveryTag = message.getMessageProperties().getDeliveryTag() ; 
  14.   byte[] buf =  null ; 
  15.   try { 
  16.    buf = message.getBody() ; 
  17.    logger.info("接受到消息：{}", new String(buf, "UTF-8")) ; 
  18.    Map<String, Object> result = new JsonMapper().readValue(buf, Map.class) ; 
  19.    Long id = ((Integer) result.get("accountId")) + 0L ; 
  20.    String orderId = (String) result.get("orderId") ; 
  21.    usersService.updateMoneyAndLogStatus(id, orderId) ; 
  22.    channel.basicAck(deliveryTag, true) ; 
  23.   } catch (Exception e) { 
  24.    logger.error("消息接受出现异常：{}, 异常消息：{}", e.getMessage(), new String(buf, Charset.forName("UTF-8"))) ; 
  25.    e.printStackTrace() ; 
  26.    try { 
  27.     // 应该将这类异常的消息放入死信队列中，以便人工排查。 
  28.     channel.basicReject(deliveryTag, false); 
  29.    } catch (IOException e1) { 
  30.     logger.error("拒绝消息重入队列异常：{}", e1.getMessage()) ; 
  31.     e1.printStackTrace(); 
  32.    } 
  33.   } 
  34.  } 
  35. } 

Controller接口

  1. @RestController 
  2. @RequestMapping("/users") 
  3. public class UsersController { 
  4.      
  5.     @Resource 
  6.     private RestTemplate restTemplate ; 
  7.     @Resource 
  8.     private UsersService usersService ; 
  9.      
  10.     @PostMapping("/pay") 
  11.     public Object pay(Long id, BigDecimal money) throws Exception { 
  12.         HttpHeaders headers = new HttpHeaders() ; 
  13.         headers.setContentType(MediaType.APPLICATION_JSON) ; 
  14.         String orderId = UUID.randomUUID().toString().replaceAll("-", "") ; 
  15.         Map<String, String> params = new HashMap<>() ; 
  16.         params.put("accountId", String.valueOf(id)) ; 
  17.         params.put("orderId", orderId) ; 
  18.         params.put("money", money.toString()) ; 
  19.          
  20.         UsersLog usersLog = new UsersLog() ; 
  21.         usersLog.setCreateTime(new Date()) ; 
  22.         usersLog.setOrderId(orderId); 
  23.         usersLog.setMoney(money) ; 
  24.         usersLog.setStatus(0) ; 
  25.         usersService.saveLog(usersLog) ; 
  26.         HttpEntity<String> requestEntity = new HttpEntity<String>(new ObjectMapper().writeValueAsString(params), headers) ; 
  27.         return restTemplate.postForObject("http://localhost:8080/payInfos/pay", requestEntity, String.class) ; 
  28.     } 
  29.      
  30. } 

以上是两个子模块的所有代码了

测试

初始数据

账户子模块控制台

支付子模块控制台

数据表数据

完毕！！！

 

文章来源网络，作者：运维，如若转载，请注明出处：https://shuyeidc.com/wp/250140.html<

赞 (0)
运维的头像运维
00
上一篇2025-04-28 09:10
下一篇 2025-04-28 09:11

相关推荐

  • 如何制作个人主题技术资讯

    个人主题怎么制作？

    制作个人主题是一个将个人风格、兴趣或专业领域转化为视觉化或结构化内容的过程，无论是用于个人博客、作品集、社交媒体账号还是品牌形象，核心都是围绕“个人特色”展开，以下从定位、内容规划、视觉设计、技术实现四个维度,详细拆解制作个人主题的完整流程，明确主题定位：找到个人特色的核心主题定位是所有工作的起点，需要先回答……

    运维的头像运维
    2025-11-20
    0
  • 如何管理社群营销技术资讯

    社群营销管理关键是什么？

    社群营销的核心在于通过建立有温度、有价值、有归属感的社群，实现用户留存、转化和品牌传播，其管理需贯穿“目标定位-内容运营-用户互动-数据驱动-风险控制”全流程,以下从五个维度展开详细说明：明确社群定位与目标社群管理的首要任务是精准定位，需明确社群的核心价值（如行业交流、产品使用指导、兴趣分享等）、目标用户画像……

    运维的头像运维
    2025-11-20
    0
  • 香港公司如何网站备案技术资讯

    香港公司网站备案需要什么材料？

    香港公司进行网站备案是一个涉及多部门协调、流程相对严谨的过程，尤其需兼顾中国内地与香港两地的监管要求，由于香港公司注册地与中国内地不同，其网站若主要服务内地用户或使用内地服务器，需根据服务器位置、网站内容性质等，选择对应的备案路径（如工信部ICP备案或公安备案），以下从备案主体资格、流程步骤、材料准备、注意事项……

    运维的头像运维
    2025-11-20
    0
  • 如何企业上云推广技术资讯

    如何企业上云推广

    企业上云已成为数字化转型的核心战略,但推广过程中需结合行业特性、企业痛点与市场需求，构建系统性、多维度的推广体系，以下从市场定位、策略设计、执行落地及效果优化四个维度，详细拆解企业上云推广的实践路径，精准定位：明确目标企业与核心价值企业上云并非“一刀切”的方案，需先锁定目标客户群体，提炼差异化价值主张，客户分层……

    运维的头像运维
    2025-11-20
    0
  • PS如何设计搜索框技术资讯

    PS设计搜索框的实用技巧有哪些？

    在PS中设计一个美观且功能性的搜索框需要结合创意构思、视觉设计和用户体验考量，以下从设计思路、制作步骤、细节优化及交互预览等方面详细说明，帮助打造符合需求的搜索框，设计前的规划明确使用场景：根据网站或APP的整体风格确定搜索框的调性，例如极简风适合细线条和纯色，科技感适合渐变和发光效果，电商类则可能需要突出搜索……

    运维的头像运维
    2025-11-20
    0

发表回复

您的邮箱地址不会被公开。必填项已用 * 标注