文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

SpringBoot结合RabbitMQ实现分布式事务之最大努力通知

2024-11-30 09:13

关注

什么是最大努力通知

这是一个充值的案例

图片

交互流程 :

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

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

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

账户系统未接收到通知会主动调用充值系统的接口查询充值结果。通过上边的例子我们总结最大努力通知方案的目标 :目标 :发起通知方通过一定的机制最大努力将业务处理结果通知到接收方。

具体包括 :

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

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

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

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

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

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

通过RabbitMQ实现最大努力通知

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

图片

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


 org.springframework.boot
 spring-boot-starter-data-jpa


 org.springframework.boot
 spring-boot-starter-web


 org.springframework.boot
 spring-boot-starter-amqp


 mysql
 mysql-connector-java
 runtime

子模块pay-manager

server:
  port: 8080
---
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        concurrency: 5
        maxConcurrency: 10
        prefetch: 5
        acknowledgeMode: MANUAL
        retry:
          enabled: true
          initialInterval: 3000
          maxAttempts: 3
        defaultRequeueRejected: false

记录充值金额及账户信息

@Entity
@Table(name = "t_pay_info")
public class PayInfo implements Serializable{
 @Id
 private Long id;
 private BigDecimal money ;
 private Long accountId ;
}
public interface PayInfoRepository extends JpaRepository {
 PayInfo findByOrderId(String orderId) ;
}
@Service
public class PayInfoService {
  
  @Resource
  private PayInfoRepository payInfoRepository ;
  @Resource
  private RabbitTemplate rabbitTemplate ;
  
  // 数据保存完后发送消息(这里发送消息可以应用确认模式或事物模式)
  @Transactional
  public PayInfo savePayInfo(PayInfo payInfo) {
    payInfo.setId(System.currentTimeMillis()) ;
    PayInfo result = payInfoRepository.save(payInfo) ;
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString().replaceAll("-", "")) ;
    try {
      rabbitTemplate.convertAndSend("pay-exchange", "pay.#", new ObjectMapper().writeValueAsString(payInfo), correlationData) ;
    } catch (AmqpException | JsonProcessingException e) {
      e.printStackTrace();
    }
    return result ;
  }
  
  public PayInfo queryByOrderId(String orderId) {
    return payInfoRepository.findByOrderId(orderId) ;
  }
  
}

支付完成后发送消息。

@RestController
@RequestMapping("/payInfos")
public class PayInfoController {
 @Resource
 private PayInfoService payInfoService ;
  
  // 支付接口
 @PostMapping("/pay")
 public Object pay(@RequestBody PayInfo payInfo) {
  payInfoService.savePayInfo(payInfo) ;
  return "支付已提交,等待结果" ;
 }
  
 @GetMapping("/queryPay")
 public Object queryPay(String orderId) {
  return payInfoService.queryByOrderId(orderId) ;
 }
  
}

子模块users-manager

server:
  port: 8081
---
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        concurrency: 5
        maxConcurrency: 10
        prefetch: 5
        acknowledgeMode: MANUAL
        retry:
          enabled: true
          initialInterval: 3000
          maxAttempts: 3
        defaultRequeueRejected: false
@Entity
@Table(name = "t_users")
public class Users {
 @Id
 private Long id;
 private String name ;
 private BigDecimal money ;
}

账户信息表

@Entity
@Table(name = "t_users_log")
public class UsersLog {
 @Id
 private Long id;
 private String orderId ;
 // 0:支付中,1:已支付,2:已取消
 @Column(columnDefinition = "int default 0")
 private Integer status = 0 ;
 private BigDecimal money ;
 private Date createTime ;
}

账户充值记录表(去重)

public interface UsersRepository extends JpaRepository {
}
public interface UsersLogRepository extends JpaRepository {
 UsersLog findByOrderId(String orderId) ;
}
@Service
public class UsersService {  
  @Resource
  private UsersRepository usersRepository ;
  @Resource
  private UsersLogRepository usersLogRepository ;
  
 @Transactional
 public boolean updateMoneyAndLogStatus(Long id, String orderId) {
  UsersLog usersLog = usersLogRepository.findByOrderId(orderId) ;
  if (usersLog != null && 1 == usersLog.getStatus()) {
   throw new RuntimeException("已支付") ;
  }
  Users users = usersRepository.findById(id).orElse(null) ;
  if (users == null) {
   throw new RuntimeException("账户不存在") ;
  }
  users.setMoney(users.getMoney().add(usersLog.getMoney())) ;
  usersRepository.save(users) ;
  usersLog.setStatus(1) ;
  usersLogRepository.save(usersLog) ;
  return true ;
 }
  
 @Transactional
 public boolean saveLog(UsersLog usersLog) {
  usersLog.setId(System.currentTimeMillis()) ;
  usersLogRepository.save(usersLog) ;
  return true ;
 }
}
@Component
public class PayMessageListener {
  
 private static final Logger logger = LoggerFactory.getLogger(PayMessageListener.class) ;
  
 @Resource
 private  UsersService usersService ;
  
 @SuppressWarnings("unchecked")
 @RabbitListener(queues = {"pay-queue"})
 @RabbitHandler
 public void receive(Message message, Channel channel) {
  long deliveryTag = message.getMessageProperties().getDeliveryTag() ;
  byte[] buf =  null ;
  try {
   buf = message.getBody() ;
   logger.info("接受到消息:{}", new String(buf, "UTF-8")) ;
   Map result = new JsonMapper().readValue(buf, Map.class) ;
   Long id = ((Integer) result.get("accountId")) + 0L ;
   String orderId = (String) result.get("orderId") ;
   usersService.updateMoneyAndLogStatus(id, orderId) ;
   channel.basicAck(deliveryTag, true) ;
  } catch (Exception e) {
   logger.error("消息接受出现异常:{}, 异常消息:{}", e.getMessage(), new String(buf, Charset.forName("UTF-8"))) ;
   e.printStackTrace() ;
   try {
    // 应该将这类异常的消息放入死信队列中,以便人工排查。
    channel.basicReject(deliveryTag, false);
   } catch (IOException e1) {
    logger.error("拒绝消息重入队列异常:{}", e1.getMessage()) ;
    e1.printStackTrace();
   }
  }
 }
}
@RestController
@RequestMapping("/users")
public class UsersController {
  
  @Resource
  private RestTemplate restTemplate ;
  @Resource
  private UsersService usersService ;
  
  @PostMapping("/pay")
  public Object pay(Long id, BigDecimal money) throws Exception {
    HttpHeaders headers = new HttpHeaders() ;
    headers.setContentType(MediaType.APPLICATION_JSON) ;
    String orderId = UUID.randomUUID().toString().replaceAll("-", "") ;
    Map params = new HashMap<>() ;
    params.put("accountId", String.valueOf(id)) ;
    params.put("orderId", orderId) ;
    params.put("money", money.toString()) ;
    
    UsersLog usersLog = new UsersLog() ;
    usersLog.setCreateTime(new Date()) ;
    usersLog.setOrderId(orderId);
    usersLog.setMoney(money) ;
    usersLog.setStatus(0) ;
    usersService.saveLog(usersLog) ;
    HttpEntity requestEntity = new HttpEntity(new ObjectMapper().writeValueAsString(params), headers) ;
    return restTemplate.postForObject("http://localhost:8080/payInfos/pay", requestEntity, String.class) ;
  }
  
}

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

测试

初始数据

图片

图片

账户子模块控制台

图片

支付子模块控制台

图片

数据表数据

图片

完毕!!!

来源:Spring全家桶实战案例源码内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯