异步通讯异步调用同步调用(同步通讯)调用者发起请求后需要等待服务提供者执行业务返回结果后才能继续执行后面的业务也就是说调用者在调用过程中处于阻塞状态OpenFeign 。同步通讯就如同打视频电话双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。异步通讯就如同发微信聊天双方的交互不是实时的你不需要立刻给对方回应。因此你可以多线操作同时跟多人聊天。同步调用的优势:时效性强,等待到结果后才返回同步调用的问题1.拓展性差不断添加业务导致臃肿2.性能下降最终整个业务的响应时长就是每次远程调用的执行时长之和3.级联失败当其中一个服务出现故障整个事务都会回滚要解决这些问题我们就必须用异步调用的方式来代替同步调用。异步调用异步调用方式其实就是基于消息通知的方式一般包含三个角色消息发送者投递消息的人就是原来的调用方消息Broker管理、暂存、转发消息你可以把它理解成微信服务器消息接收者接收和处理消息的人就是原来的服务提供方在异步调用中发送者不再直接同步调用接收者的业务接口而是发送一条消息投递给消息Broker。然后接收者根据自己的需求从消息Broker那里订阅消息。每当发送方发送消息后接受者都能获取消息并处理。 这样发送消息的人和接收消息的人就完全解耦了。以余额支付业务为例除了扣减余额、更新支付流水单状态以外其它调用逻辑全部取消。而是改为发送一条消息到Broker。而相关的微服务都可以订阅消息通知一旦消息到达Broker则会分发给每一个订阅了的微服务处理各自的业务。假如产品经理提出了新的需求比如要在支付成功后更新用户积分。支付代码完全不用变更而仅仅是让积分服务也订阅消息即可不管后期增加了多少消息订阅者作为支付服务来讲执行问扣减余额、更新支付流水状态后发送消息即可。业务耗时仅仅是这三部分业务耗时仅仅100ms大大提高了业务性能。另外不管是交易服务、通知服务还是积分服务他们的业务与支付关联度低。现在采用了异步调用解除了耦合他们即便执行过程中出现了故障也不会影响到支付服务。异步调用的优势耦合度更低,接触耦合,拓展性强无需等待,性能更好业务拓展性强故障隔离下游服务故障不影响上游业务,避免级联失败缓存消息,流量削峰填谷(把激增的流量放到压力较小的时候慢慢执行)流量整形异步调用的缺点不能立即得到调用结果,时效性差不确定下游业务执行是否成功完全依赖于Broker的可靠性、安全性和性能架构复杂后期维护和调试麻烦使用异步调用的场景:1.对调用结果不关心,对整个业务的影响不大(例如订单的状态更新)2.对性能要求较高,调用链很长技术选型消息Broker目前常见的实现方案就是消息队列MessageQueue简称为MQ.就是异步调用中的Broker 目比较常见的MQ实现ActiveMQRabbitMQRocketMQKafka几种常见MQ的对比RabbitMQActiveMQRocketMQKafka公司/社区RabbitApache阿里Apache开发语言Erlang面向并发的语言JavaJavaScalaJava协议支持AMQPXMPPSMTPSTOMPOpenWire,STOMPREST,XMPP,AMQP自定义协议自定义协议可用性高一般高高单机吞吐量(并发能力)一般差高非常高消息延迟微秒级毫秒级毫秒级毫秒以内消息可靠性(消息不丢失性)高一般高一般目前国内消息队列使用最多的还是RabbitMQ再加上其各方面都比较均衡稳定性也好。RabbitMQRabbitMQ是基于Erlang语言开发的开源消息通信中间件官网地址 Messaging that just works — RabbitMQ安装基于Docker来安装RabbitMQ使用下面的命令即可docker run \ -e RABBITMQ_DEFAULT_USER 用户名 \ -e RABBITMQ_DEFAULT_PASS 密码 \ -v mq-plugins:/plugins \ --name mq \ --hostname mq \ -p 15672:15672 \ -p 5672:5672 \ --network hmall \ -d \ rabbitmq:3.8-management以看到在安装命令中有两个映射的端口15672RabbitMQ提供的管理控制台的端口5672RabbitMQ的消息发送处理接口访问 http://192.168.xxx.xxx:15672(IP端口号)即可看到管理控制台。首次访问需要登录默认的用户名和密码在配置文件中已经指定了。登录后即可看到管理控制台总览页面RabbitMQ对应的架构如图其中包含几个概念publisher生产者也就是发送消息的一方consumer消费者也就是消费消息的一方queue队列存储消息。生产者投递的消息会暂存在消息队列中等待消费者处理exchange交换机负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。交换机负责路由和转发消息,没有存储消息的能力virtual host虚拟主机起到数据隔离的作用。每个虚拟主机相互独立有各自的exchange、queue收发消息1 发送到交换机的消息只会路由到与其绑定的队列因此仅仅创建队列是不够的还需要将其与交换机绑定2 绑定关系:点击Exchanges选项卡点击amq.fanout交换机进入交换机详情页然后点击Bindings菜单在表单中填写要绑定的队列名称3 再次回到exchange页面找到刚刚绑定的amq.fanout点击进入详情页再次发送一条消息(在Payload中填写消息体)4 点击队列名称进入详情页查看队列详情这次我们点击get message,可以看到消息到达队列注意点1 如果交换机和队列没有绑定,直接向交换机发送消息不能成功(交换机没有存储消息的能力)2 多个队列绑定同一个交换机,发送的消息多个队列都能收到数据隔离用户管理这里的用户都是RabbitMQ的管理或运维人员。目前只有安装RabbitMQ时添加的itheima这个用户。仔细观察用户表格中的字段如下Nameitheima也就是用户名Tagsadministrator说明itheima用户是超级管理员拥有所有权限Can access virtual host/可以访问的virtual host这里的/是默认的virtual host对于小型企业而言出于成本考虑通常只会搭建一套MQ集群公司内的多个不同项目同时使用。这个时候为了避免互相干扰 利用virtual host的隔离特性将不同项目隔离(不同虚拟主机的用户名可以相同 )。一般会做两件事情给每个项目创建独立的运维账号将管理权限分离。给每个项目创建不同的virtual host将每个项目的数据隔离。SpringAMQPAMQPAdvanced Message Queuing Protocol是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关更符合微服务中独立性的要求。Spring AMQPSpring AMQP是基于AMQP协议定义的一套API规范提供了模板来发送和接收消息。包含两部分其中spring-amqp是基础抽象spring-rabbit是底层的默认实现。开发业务功能的时候是应该基于编程的方式收发消息。由于RabbitMQ采用了AMQP协议因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。但是RabbitMQ官方提供的Java客户端编码相对复杂一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具SpringAMQP。并且还基于SpringBoot对其实现了自动装配使用起来非常方便。SpringAMQP提供了三个功能自动声明队列、交换机及其绑定关系基于注解的监听器模式异步接收消息封装了RabbitTemplate工具用于发送消息快速入门测试使用一般都是经过交换机发送消息到队列不过有时候为了测试方便我们也可以直接向队列发送消息跳过交换机。publisher直接发送消息到队列(少一个交换机来路由和转发)消费者监听并处理队列中的消息1.先在控制台新建一个队列simple.queue2.消息发送:首先配置MQ地址:spring: rabbitmq: host: 192.168.xxx.xxx # 你的虚拟机IP port: 5672 # 端口 virtual-host: /hmall # 虚拟主机 username: hmall # 用户名 password: 123 # 密码编写测试类SpringAmqpTest,利用RabbitTemplate实现消息发送SpringBootTest public class SpringAmqpTest { Autowired private RabbitTemplate rabbitTemplate; Test public void testSimpleQueue() { // 队列名称 String queueName simple.queue; // 消息 String message hello, spring amqp!; // 发送消息 message是Object类,可以是任何类型 rabbitTemplate.convertAndSend(queueName, message); } }3.消息接收先配置MQ地址,和上面一样新建一个类SpringRabbitListener代码如下Component public class SpringRabbitListener { // 利用RabbitListener来声明要监听的队列信息 // 将来一旦监听的队列中有了消息就会推送给当前服务调用当前方法处理消息。 // 可以看到方法体中接收的就是消息体的内容 RabbitListener(queues simple.queue) public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println(spring 消费者接收到消息【 msg 】); } }SpringAMQP如何收发消息① 引入spring-boot-starter-amqp依赖② 配置rabbitmq服务端信息③ 利用RabbitTemplate发送消息④ 利用RabbitListener注解声明要监听的队列监听消息WorkQueues模型Work queues任务模型。简单来说就是让多个消费者绑定到一个队列共同消费队列中的消息。当消息处理比较耗时的时候可能生产消息的速度会远远大于消息的消费速度。长此以往消息就会堆积越来越多无法及时处理。 使用work 模型多个消费者共同处理消息处理消息处理的速度就能大大提高。能者多劳:1.创建新队列命名为work.queue2.添加配置:spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息若不添加Prefetch为1,则多个消费者绑定同一个队列,队列投递消息类似轮询机制,消息是平均分配给每个消费者并没有考虑到消费者的处理能力。导致1个消费者空闲另一个消费者忙的不可开交。2.消息发送:在publisher服务中的SpringAmqpTest类中添加一个测试方法Test public void testWorkQueue() throws InterruptedException { // 队列名称 String queueName simple.queue; // 消息 String message hello, message_; for (int i 0; i 50; i) { // 发送消息每20毫秒发送一次相当于每秒发送50条消息 rabbitTemplate.convertAndSend(queueName, message i); Thread.sleep(20); } }3.消息接收要模拟多个消费者绑定同一个队列我们在consumer服务的SpringRabbitListener中添加2个新的方法RabbitListener(queues work.queue) public void listenWorkQueue1(String msg) throws InterruptedException { System.out.println(消费者1接收到消息【 msg 】 LocalTime.now()); Thread.sleep(20); } RabbitListener(queues work.queue) public void listenWorkQueue2(String msg) throws InterruptedException { // err红色打印 System.err.println(消费者2........接收到消息【 msg 】 LocalTime.now()); Thread.sleep(200); }注意到这两消费者都设置了Thead.sleep模拟任务耗时消费者1 sleep了20毫秒相当于每秒钟处理50个消息消费者2 sleep了200毫秒相当于每秒处理5个消息由于消费者1处理速度较快所以处理了更多的消息消费者2处理速度较慢只处理了6条消息。而最终总的执行耗时也在1秒左右大大提升。 正所谓能者多劳这样充分利用了每一个消费者的处理能力可以有效避免消息积压问题。Work模型的使用多个消费者绑定到一个队列可以加快消息处理速度(解决消息堆积问题)同一条消息只会被一个消费者处理通过设置prefetch: 1来控制消费者预取的消息数量,处理完一条再处理下一条,实现能者多劳交换机类型在订阅模型中多了一个exchange角色而且过程略有变化Publisher生产者不再发送消息到队列中而是发给交换机Exchange交换机一方面接收生产者发送的消息。另一方面知道如何处理消息例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作取决于Exchange的类型。Queue消息队列也与以前一样接收消息、缓存消息。不过队列一定要与交换机绑定。Consumer消费者与以前一样订阅队列没有变化Exchange交换机只负责转发消息不具备存储消息的能力因此如果没有任何队列与Exchange绑定或者没有符合路由规则的队列那么消息会丢失交换机的类型有四种Fanout广播将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机Direct订阅基于RoutingKey路由key发送给订阅了消息的队列Topic通配符订阅与Direct类似只不过RoutingKey可以使用通配符Headers头匹配基于MQ的消息头匹配用的较少。Fanout交换机(广播模式)1 可以有多个队列2 每个队列都要绑定到Exchange交换机3 生产者发送的消息只能发送到交换机4 交换机把消息发送给绑定过的所有队列5 订阅队列的消费者都能拿到消息使用1.创建一个名为hmall.fanout的交换机类型是Fanout2.创建两个队列fanout.queue1和fanout.queue2绑定到交换机hmall.fanout3.消息发送:在publisher服务的SpringAmqpTest类中添加测试方法Test public void testFanoutExchange() { // 交换机名称 String exchangeName hmall.fanout; // 消息 String message hello, everyone!; rabbitTemplate.convertAndSend(exchangeName, , message); }4.消息接收在consumer服务的SpringRabbitListener中添加两个方法作为消费者RabbitListener(queues fanout.queue1) public void listenFanoutQueue1(String msg) { System.out.println(消费者1接收到Fanout消息【 msg 】); } RabbitListener(queues fanout.queue2) public void listenFanoutQueue2(String msg) { System.out.println(消费者2接收到Fanout消息【 msg 】); }在Fanout模式中一条消息会被所有订阅的队列都消费。交换机的作用是什么接收publisher发送的消息将消息按照规则路由到与之绑定的队列不能缓存消息路由失败消息丢失FanoutExchange的会将消息路由到每个绑定的队列Direct交换机在某些场景下希望不同的消息被不同的队列消费要用到Direct类型的Exchange。在Direct模型下每一个Queue都与Exchange设置一个BindingKey队列与交换机的绑定不能是任意绑定了而是要指定一个BindingKey路由key消息的发送方在 向 Exchange发送消息时也必须指定消息的RoutingKey。Exchange不再把消息交给每一个绑定的队列而是根据消息的Routing Key进行判断只有队列的BindingKey与消息的Routing key完全一致才会接收到消息消息接收RabbitListener(queues direct.queue1) public void listenDirectQueue1(String msg) { System.out.println(消费者1接收到direct.queue1的消息【 msg 】); }消息发送Test public void testSendDirectExchange() { // 交换机名称 String exchangeName hmall.direct; // 消息 String message 今天是7月8号; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, red, message); }由于使用的red这个key所以只有BindingKey为red的队列的消费者才能收到消息Direct交换机与Fanout交换机的差异Fanout交换机将消息路由给每一个与之绑定的队列Direct交换机根据RoutingKey判断路由给哪个队列如果多个队列具有相同的RoutingKey则与Fanout功能类似Topic交换机Topic类型的Exchange与Direct相比都是可以根据RoutingKey把消息路由到不同的队列。 只不过Topic类型Exchange可以让队列在绑定BindingKey的时候使用通配符BindingKey一般都是有一个或多个单词组成多个单词之间以.分割例如item.insert通配符规则#匹配一个或多个词*匹配不多不少恰好1个词举例item.#能够匹配item.spu.insert或者item.spuitem.*只能匹配item.spuDirect交换机与Topic交换机的差异Topic交换机接收的消息RoutingKey必须是多个单词以**.**分割Topic交换机与队列绑定时的bindingKey可以指定通配符#代表0个或多个词*代表1个词声明队列和交换机由程序启动时检查队列和交换机是否存在如果不存在自动创建。SpringAMQP 提供了几个类用来声明队列、交换机及其绑定关系Queue用于声明队列可以用工厂类 QueueBuilder 构建Exchange用于声明交换机可以用工厂类 ExchangeBuilder 构建Binding用于声明队列和交换机的绑定关系可以用工厂类 BindingBuilder 构建三大对象的作用Queue队列消息最终存储载体接收交换机转发的消息消费者从队列拉取消息QueueBuilder 支持快速配置队列持久化、是否自动删除、专属队列等属性。Exchange交换机消息分发中转站生产者只发送消息到交换机交换机根据路由规则转发至对应队列ExchangeBuilder 可构建 Direct/Fanout/Topic/Headers 四种类型交换机。Binding绑定建立交换机与队列的关联关系绑定中指定路由 key决定交换机消息的分发目标BindingBuilder 链式绑定交换机、队列、路由键。使用方法在 SpringBoot 项目中将这三类对象注册为BeanSpringAMQP 会在项目启动时自动连接 RabbitMQ完成队列、交换机、绑定关系的创建不存在则创建已存在则跳过不会覆盖原有配置。fanout示例在consumer中创建一个类声明队列和交换机Configuration public class FanoutConfig { /** * 声明交换机 * return Fanout类型交换机 */ //所有加Bean的都会被Spring动态代理,默认会先检查Spring容器里有没有这个Bean,FanoutExchange是类型,fanoutExchange是方法名称 //如果有就直接返回这个Bean,如果没有才会执行这个方法生成Bean Bean public FanoutExchange fanoutExchange(){ // 或者ExchangeBuilder.fanoutExchange(hmall.fanout).build(); return new FanoutExchange(hmall.fanout); } /** * 第1个队列 */ Bean public Queue fanoutQueue1(){ // 或者QueueBuilder.durable(fanout.queue1).build(); return new Queue(fanout.queue1);//默认durable持久 } /** * 绑定队列和交换机 */ Bean public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } /** * 第2个队列 */ Bean public Queue fanoutQueue2(){ return new Queue(fanout.queue2); } /** * 绑定队列和交换机 */ Bean public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); } }Bean public Queue fanoutQueue3(){ return new Queue(fanout.queue3); } Bean public Binding fanoutBinding3(Queue fanoutQueue3, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue3).to(fanoutExchange); } Bean public Queue fanoutQueue4(){ return new Queue(fanout.queue4); } Bean public Binding fanoutBinding4(){ return BindingBuilder.bind(fanoutQueue4()).to(fanoutExchange()); }两种注入 Bean 的方式fanoutBinding3通过方法参数自动注入容器中已存在的Queue、FanoutExchangeBeanSpring 自动完成依赖装配推荐使用。fanoutBinding4直接调用对应 Bean 的创建方法fanoutQueue3()、fanoutExchange()来获取实例同样能完成绑定。Fanout 交换机特性Fanout 类型交换机绑定时不需要指定路由 key所有绑定该交换机的队列都会收到消息因此链式调用末尾无需拼接.with(路由key)。队列创建new Queue(fanout.queue4)默认创建持久化、非独占、不自动删除的队列direct示例direct模式由于要绑定多个KEY会非常麻烦每一个Key都要编写一个bindingConfiguration 配置 public class DirectConfig { /** * 声明交换机 * return Direct类型交换机 */ Bean public DirectExchange directExchange(){ return ExchangeBuilder.directExchange(hmall.direct).build(); } /** * 第1个队列 */ Bean public Queue directQueue1(){ return new Queue(direct.queue1); } /** * 绑定队列和交换机 */ Bean public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){ return BindingBuilder.bind(directQueue1).to(directExchange).with(red); } /** * 绑定队列和交换机 */ Bean public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){ return BindingBuilder.bind(directQueue1).to(directExchange).with(blue);基于注解声明基于Bean的方式声明队列和交换机比较麻烦Spring还提供了基于注解方式来声明。例如我们同样声明Direct模式的交换机和队列Ctrlp有注解提示RabbitListener(bindings QueueBinding( value Queue(name direct.queue1), exchange Exchange(name hmall.direct, type ExchangeTypes.DIRECT), key {red, blue} )) public void listenDirectQueue1(String msg){ System.out.println(消费者1接收到direct.queue1的消息【 msg 】); } RabbitListener(bindings QueueBinding( value Queue(name direct.queue2), exchange Exchange(name hmall.direct, type ExchangeTypes.DIRECT), key {red, yellow} )) public void listenDirectQueue2(String msg){ System.out.println(消费者2接收到direct.queue2的消息【 msg 】); }Topic模式RabbitListener(bindings QueueBinding( value Queue(name topic.queue1), exchange Exchange(name hmall.topic, type ExchangeTypes.TOPIC), key china.# )) public void listenTopicQueue1(String msg){ System.out.println(消费者1接收到topic.queue1的消息【 msg 】); } RabbitListener(bindings QueueBinding( value Queue(name topic.queue2), exchange Exchange(name hmall.topic, type ExchangeTypes.TOPIC), key #.news )) public void listenTopicQueue2(String msg){ System.out.println(消费者2接收到topic.queue2的消息【 msg 】); }总结一 声明队列、交换机、绑定关系的 Bean 对象Queue队列对象用于定义消息存储队列交换机细分类型FanoutExchange广播交换机DirectExchange直连交换机TopicExchange主题交换机Binding绑定对象建立队列与交换机的关联关系使用方式将以上对象注册为Bean项目启动时自动在 RabbitMQ 创建对应资源。二 RabbitListener 配套声明资源的注解Queue注解式定义队列Exchange 注解式定义交换机实际开发中还会搭配 QueueBinding方式优点缺点Bean 声明资源统一管理生产者、消费者共用解耦代码分散在配置类RabbitListener 注解内嵌声明消费者和资源定义写在一起直观生产者无法复用易重复定义消息转换器Spring的消息发送代码接收的消息体是一个Object而在数据传输时它会把你发送的消息序列化为字节码格式发送给MQ接收消息的时候还会把字节反序列化为Java对象。 只不过默认情况下Spring采用的序列化方式是JDK序列化。需要配置JSON转换器。JDK序列化存在下列问题数据体积过大有安全漏洞可读性差配置JSON转换器显然JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高因此可以使用JSON方式来做序列化和反序列化。在publisher和consumer两个服务中都引入依赖dependency groupIdcom.fasterxml.jackson.dataformat/groupId artifactIdjackson-dataformat-xml/artifactId version2.9.10/version /dependency如果项目中引入了spring-boot-starter-web依赖则无需再次引入Jackson依赖。配置消息转换器在publisher和consumer两个服务的启动类中添加一个BeanBean public MessageConverter messageConverter(){ // 1.定义消息转换器 Jackson2JsonMessageConverter jackson2JsonMessageConverter new Jackson2JsonMessageConverter(); // 2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息 jackson2JsonMessageConverter.setCreateMessageIds(true); return jackson2JsonMessageConverter; }将业务的同步调用改为异步调用(以topic交换机为例)改造余额支付功能将支付成功后基于OpenFeign的交易服务的更新订单状态接口的同步调用改为基于RabbitMQ的异步通知。我们只关注交易服务步骤如下定义topic类型交换机命名为pay.topic定义消息队列命名为mark.order.pay.queue将mark.order.pay.queue与pay.topic绑定BindingKey为pay.success支付成功时不再调用交易服务更新订单状态的接口而是发送一条消息到pay.topic发送消息的RoutingKey为pay.success消息内容是订单id交易服务监听mark.order.pay.queue队列接收到消息后更新订单状态为已支付1.配置MQ不管是生产者还是消费者都需要配置MQ的基本信息。分为两步1添加依赖!--消息发送-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency2配置MQ地址spring: rabbitmq: host: 192.168.xxx.xxx # 你的虚拟机IP port: 5672 # 端口 virtual-host: /hmall # 虚拟主机 username: hmall # 用户名 password: 123 # 密码2.接收消息在listener中定义一个消息监听类Component RequiredArgsConstructor public class PayStatusListener { private final IOrderService orderService; // 注解 RabbitListener(bindings QueueBinding( value Queue(name mark.order.pay.queue, durable true), exchange Exchange(name pay.topic, type ExchangeTypes.TOPIC), key pay.success )) public void listenPaySuccess(Long orderId){ orderService.markOrderPaySuccess(orderId); } }3.发送消息private final RabbitTemplate rabbitTemplate; Override Transactional public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) { // 1.查询支付单 PayOrder po getById(payOrderDTO.getId()); // 2.判断状态 if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){ // 订单不是未支付状态异常 throw new BizIllegalException(交易已支付或关闭); } // 3.尝试扣减余额 userClient.deductMoney(payOrderDTO.getPw(), po.getAmount()); // 4.修改支付单状态 boolean success markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now()); if (!success) { throw new BizIllegalException(交易已支付或关闭); } // 5.修改订单状态 // tradeClient.markOrderPaySuccess(po.getBizOrderNo()); try { rabbitTemplate.convertAndSend(pay.topic, pay.success, po.getBizOrderNo()); } catch (Exception e) {//避免出现异常导致事务回滚 log.error(支付成功的消息发送失败支付单id{} 交易单id{}, po.getId(), po.getBizOrderNo(), e); } }rabbitTemplate.convertAndSend(pay.topic, pay.success, po.getBizOrderNo());