使用rabbitmq进行支付之后的消息通知

news2025/1/24 14:35:43

订单服务完成支付后将支付结果发给每一个与订单服务对接的微服务，订单服务将消息发给交换机，由交换机广播消息，每个订阅消息的微服务都可以接收到支付结果.

微服务收到支付结果根据订单的类型去更新自己的业务数据。

相关技术方案

使用消息队列进行异步通知需要保证消息的可靠性，即生产端将消息成功通知到消费端。

消息从生产端发送到消费端经历了如下过程：

1、消息发送到交换机

2、消息由交换机发送到队列

3、消息者收到消息进行处理

保证消息的可靠性需要保证以上过程的可靠性，本项目使用RabbitMQ可以通过如下方面保证消息的可靠性。

1、生产者确认机制

发送消息前使用数据库事务将消息保证到数据库表中

成功发送到交换机将消息从数据库中删除

2、mq持久化

mq收到消息进行持久化，当mq重启即使消息没有消费完也不会丢失。

需要配置交换机持久化、队列持久化、发送消息时设置持久化。

3、消费者确认机制

消费者消费成功自动发送ack，否则重试消费。

订单服务通过消息队列将支付结果发给学习中心服务，消息队列采用发布订阅模式。

1、订单服务创建支付结果通知交换机。

2、学习中心服务绑定队列到交换机。

首先需要在学习中心服务和订单服务工程配置连接消息队列。

1、首先在订单服务添加消息队列依赖

XML

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、在nacos配置rabbitmq-dev.yaml为通用配置文件

YAML
spring:
  rabbitmq:
    host: 192.168.101.65
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-confirm-type: correlated #correlated 异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
    publisher-returns: false #开启publish-return功能，同样是基于callback机制，需要定义ReturnCallback
    template:
      mandatory: false #定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息
    listener:
      simple:
        acknowledge-mode: none #出现异常时返回unack，消息回滚到mq；没有异常，返回ack ,manual:手动控制,none:丢弃消息，不回滚到mq
        retry:
          enabled: true #开启消费者失败重试
          initial-interval: 1000ms #初识的失败等待时长为1秒
          multiplier: 1 #失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 #最大重试次数
          stateless: true #true无状态；false有状态。如果业务中包含事务，这里改为false

3、在订单服务接口工程引入rabbitmq-dev.yaml配置文件

YAML
shared-configs:
  - data-id: rabbitmq-${spring.profiles.active}.yaml
    group: xuecheng-plus-common
    refresh: true

4、在订单服务service工程编写MQ配置类，配置交换机

Java
package com.xuecheng.orders.config;

import com.alibaba.fastjson.JSON;
import com.xuecheng.messagesdk.model.po.MqMessage;
import com.xuecheng.messagesdk.service.MqMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author Mr.M
 * @version 1.0
 * @description TODO
 * @date 2023/2/23 16:59
 */
@Slf4j
@Configuration
public class PayNotifyConfig implements ApplicationContextAware {

    //交换机
    public static final String PAYNOTIFY_EXCHANGE_FANOUT = "paynotify_exchange_fanout";
    //支付结果通知消息类型
    public static final String MESSAGE_TYPE = "payresult_notify";
    //支付通知队列
    public static final String PAYNOTIFY_QUEUE = "paynotify_queue";

    //声明交换机，且持久化
    @Bean(PAYNOTIFY_EXCHANGE_FANOUT)
    public FanoutExchange paynotify_exchange_fanout() {
        // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
        return new FanoutExchange(PAYNOTIFY_EXCHANGE_FANOUT, true, false);
    }
    //支付通知队列,且持久化
    @Bean(PAYNOTIFY_QUEUE)
    public Queue course_publish_queue() {
        return QueueBuilder.durable(PAYNOTIFY_QUEUE).build();
    }

    //交换机和支付通知队列绑定
    @Bean
    public Binding binding_course_publish_queue(@Qualifier(PAYNOTIFY_QUEUE) Queue queue, @Qualifier(PAYNOTIFY_EXCHANGE_FANOUT) FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 获取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        //消息处理service
        MqMessageService mqMessageService = applicationContext.getBean(MqMessageService.class);
        // 设置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 投递失败，记录日志
            log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
                    replyCode, replyText, exchange, routingKey, message.toString());
            MqMessage mqMessage = JSON.parseObject(message.toString(), MqMessage.class);
            //将消息再添加到消息表
            mqMessageService.addMessage(mqMessage.getMessageType(),mqMessage.getBusinessKey1(),mqMessage.getBusinessKey2(),mqMessage.getBusinessKey3());

        });
    }
}

重启订单服务，登录rabbitmq，查看交换机自动创建成功

查看队列自动成功

4.3.2 发送支付结果

在OrderService中定义接口

Java
/**
 * 发送通知结果
 * @param message
 */
public void notifyPayResult(MqMessage message);

编写接口实现方法：

Java
@Override
public void notifyPayResult(MqMessage message) {

    //1、消息体，转json
    String msg = JSON.toJSONString(message);
    //设置消息持久化
    Message msgObj = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .build();
    // 2.全局唯一的消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(message.getId().toString());
    // 3.添加callback
    correlationData.getFuture().addCallback(
            result -> {
                if(result.isAck()){
                    // 3.1.ack，消息成功
                    log.debug("通知支付结果消息发送成功, ID:{}", correlationData.getId());
                    //删除消息表中的记录
                    mqMessageService.completed(message.getId());
                }else{
                    // 3.2.nack，消息失败
                    log.error("通知支付结果消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                }
            },
            ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 发送消息
    rabbitTemplate.convertAndSend(PayNotifyConfig.PAYNOTIFY_EXCHANGE_FANOUT, "", msgObj,correlationData);

}

订单服务收到第三方平台的支付结果时，在saveAliPayStatus方法中添加代码，向数据库消息表添加消息并进行发送消息，如下所示：

Java
@Transactional
@Override
public void saveAliPayStatus(PayStatusDto payStatusDto) {
        .......
        //保存消息记录,参数1：支付结果通知类型，2: 业务id，3:业务类型
        MqMessage mqMessage = mqMessageService.addMessage("payresult_notify", orders.getOutBusinessId(), orders.getOrderType(), null);
        //通知消息
        notifyPayResult(mqMessage);
    }
}

配置交换机和队列

在order-service工程配置

消息发送方法

Java
/**
 * 发送通知结果
 * @param message
 */
public void notifyPayResult(MqMessage message);
 

4.4 接收支付结果

4.4.1 学习中心服务集成MQ

1、在学习中心服务添加消息队列依赖

XML
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、在学习中心服务接口工程引入rabbitmq-dev.yaml配置文件

YAML
shared-configs:
  - data-id: rabbitmq-${spring.profiles.active}.yaml
    group: xuecheng-plus-common
    refresh: true

3、添加配置类

Java
package com.xuecheng.learning.config;

import com.alibaba.fastjson.JSON;
import com.xuecheng.messagesdk.model.po.MqMessage;
import com.xuecheng.messagesdk.service.MqMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author Mr.M
 * @version 1.0
 * @description TODO
 * @date 2023/2/23 16:59
 */
@Slf4j
@Configuration
public class PayNotifyConfig {

    //交换机
    public static final String PAYNOTIFY_EXCHANGE_FANOUT = "paynotify_exchange_fanout";
    //支付结果通知消息类型
    public static final String MESSAGE_TYPE = "payresult_notify";
    //支付通知队列
    public static final String PAYNOTIFY_QUEUE = "paynotify_queue";

    //声明交换机，且持久化
    @Bean(PAYNOTIFY_EXCHANGE_FANOUT)
    public FanoutExchange paynotify_exchange_fanout() {
        // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
        return new FanoutExchange(PAYNOTIFY_EXCHANGE_FANOUT, true, false);
    }
    //支付通知队列,且持久化
    @Bean(PAYNOTIFY_QUEUE)
    public Queue course_publish_queue() {
        return QueueBuilder.durable(PAYNOTIFY_QUEUE).build();
    }

    //交换机和支付通知队列绑定
    @Bean
    public Binding binding_course_publish_queue(@Qualifier(PAYNOTIFY_QUEUE) Queue queue, @Qualifier(PAYNOTIFY_EXCHANGE_FANOUT) FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }

}

4.4.2 接收支付结果

监听MQ，接收支付结果，定义ReceivePayNotifyService类如下：

Java
package com.xuecheng.learning.service.impl;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.xuecheng.base.exception.XueChengPlusException;
import com.xuecheng.learning.config.PayNotifyConfig;
import com.xuecheng.learning.service.MyCourseTablesService;
import com.xuecheng.messagesdk.model.po.MqMessage;
import com.xuecheng.messagesdk.service.MqMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * @author Mr.M
 * @version 1.0
 * @description 接收支付结果
 * @date 2023/2/23 19:04
 */
@Slf4j
@Service
public class ReceivePayNotifyService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    MqMessageService mqMessageService;

    @Autowired
    MyCourseTablesService myCourseTablesService;


    //监听消息队列接收支付结果通知
    @RabbitListener(queues = PayNotifyConfig.PAYNOTIFY_QUEUE)
    public void receive(Message message, Channel channel) {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        //获取消息
        MqMessage mqMessage = JSON.parseObject(message.getBody(), MqMessage.class);
        log.debug("学习中心服务接收支付结果:{}", mqMessage);

        //消息类型
        String messageType = mqMessage.getMessageType();
        //订单类型,60201表示购买课程
        String businessKey2 = mqMessage.getBusinessKey2();
        //这里只处理支付结果通知
        if (PayNotifyConfig.MESSAGE_TYPE.equals(messageType) && "60201".equals(businessKey2)) {
            //选课记录id
            String choosecourseId = mqMessage.getBusinessKey1();
            //添加选课
            boolean b = myCourseTablesService.saveChooseCourseStauts(choosecourseId);
            if(!b){
                //添加选课失败，抛出异常，消息重回队列
                XueChengPlusException.cast("收到支付结果，添加选课失败");
            }
        }


    }


}

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【LeetCode】85.最大矩形

【LeetCode】85.最大矩形

题目 给定一个仅包含 0 和 1 、大小为 rows x cols 的二维二进制矩阵&#xff0c;找出只包含 1 的最大矩形&#xff0c;并返回其面积。 示例 1&#xff1a; 输入&#xff1a;matrix [["1","0","1","0","0"],["1&quo…
阅读更多...
11. 微积分 - 偏导数方向导数

11. 微积分 - 偏导数方向导数

文章目录 偏导数方向导数方向余弦投影继续讲方向导数Hi, 大家好。我是茶桁。 我们上节课学习了链式法则，本节课，我们要学习「偏导数」和「方向导数」。 偏导数 偏导数在导论课里面也提到过。偏导数针对多元函数去讲的。 多元函数是什么，我们拿个例子来看： 多元函数： y…
阅读更多...
springboot配置ym管理各种日记(log)

springboot配置ym管理各种日记(log)

1&#xff1a;yml配置mybatis_plus默认日记框架 mybatis-plus:#这个作用是扫描xml文件生效可以和mapper接口文件使用&#xff0c;#如果不加这个,就无法使用xml里面的sql语句#启动类加了MapperScan是扫描指定包下mapper接口生效&#xff0c;如果不用MapperScan可以在每一个mapp…
阅读更多...
2023.9.2 关于 JVM 垃圾回收机制（GC）

2023.9.2 关于 JVM 垃圾回收机制（GC）

目录 为什么要有垃圾回收机制? STW&#xff08;Stop The World&#xff09;问题 垃圾回收机制主要回收哪个内存区域? 垃圾对象判断算法 引用计数算法 可达性分析算法 垃圾对象回收算法 标记清除算法 复制算法 标记整理算法 分代算法 为什么要有垃圾回收机制? 自动…
阅读更多...
thinkphp中使用Elasticsearch 7.0进行多表的搜索

thinkphp中使用Elasticsearch 7.0进行多表的搜索

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、thinkphp中使用Elasticsearch 7.0进行多表的搜索二、使用步骤1.引入库2.读入数据 总结 前言 提示&#xff1a;thinkphp中使用Elasticsearch 7.0进行多表的…
阅读更多...
stable diffusion实践操作-批次出图

stable diffusion实践操作-批次出图

系列文章目录 stable diffusion实践操作 文章目录 系列文章目录前言一、批次出图介绍1.1 webUI设置1.2 参数介绍 二、批次出图使用2.1 如何设置2.1 效果展示 总结 前言 本章主要介绍SD批次出图。 想要一次产生多张图片的时候使用。 一、批次出图介绍 1.1 webUI设置 1.2 参数…
阅读更多...
[管理与领导-65]：IT基层管理者 - 辅助技能 - 4- 乌卡时代(VUCA )

[管理与领导-65]：IT基层管理者 - 辅助技能 - 4- 乌卡时代(VUCA )

前言&#xff1a; 大多数IT人&#xff0c;很勤奋&#xff0c;但都没有职业规划&#xff0c;被工作驱动着前行&#xff0c;然而&#xff0c;作为管理者&#xff0c;你就不能没有职业规划思维&#xff0c;因为你代表一个团队&#xff0c;你的思维决定了一个团队的思维。本文探讨…
阅读更多...
2023-9-2 染色法判定二分图

2023-9-2 染色法判定二分图

题目链接&#xff1a;染色法判定二分图 #include <iostream> #include <cstring> #include <algorithm>using namespace std;const int N 100010l, M 200010;int n, m; int h[N], e[M], ne[M], idx;int color[N];void add(int a, int b) {e[idx] b, ne[id…
阅读更多...
CSS中border-radius的来美化table的实战方案

CSS中border-radius的来美化table的实战方案

border-radius是一种CSS属性&#xff0c;用于设置元素的边框的圆角程度。其具体的用法如下&#xff1a; 设置一个值&#xff1a;可以为元素设置一个单一的圆角半径&#xff0c;这个半径将应用于元素的四个角。例如&#xff1a; div {border-radius: 10px; }设置四个值&#x…
阅读更多...
vue Cesium接入在线地图

vue Cesium接入在线地图

Cesium接入在线地图只需在创建时将imageryProvider属性换为在线地图的地址即可。 目录 天地图 OSM地图 ArcGIS 地图 谷歌影像地图 天地图 //矢量服务let imageryProvider new Cesium.WebMapTileServiceImageryProvider({url: "http://t0.tianditu.com/vec_w/wmts?s…
阅读更多...
创建性-构造者设计模式

创建性-构造者设计模式

前言 我们在使用Retrofit等这些第三方框架的时候&#xff0c;发现他们的使用都很方便&#xff0c;比如Retrofit retrofit new Retrofit.Builder().build()&#xff0c;和我们通常直接new一个对象不同&#xff0c;他是交给Builder类&#xff0c;通过build()函数来构造一个Retro…
阅读更多...
解决Ubuntu 或Debian apt-get IPv6问题：如何设置仅使用IPv4

解决Ubuntu 或Debian apt-get IPv6问题：如何设置仅使用IPv4

文章目录 解决Ubuntu 或Debian apt-get IPv6问题&#xff1a;如何设置仅使用IPv4 解决Ubuntu 或Debian apt-get IPv6问题&#xff1a;如何设置仅使用IPv4 背景&#xff1a; 在Ubuntu 22.04(包括 20.04 18.04 等版本) 或 Debian (10、11、12)系统中&#xff0c;当你使用apt up…
阅读更多...
JS中的new操作符

JS中的new操作符

文章目录 JS中的new操作符一、什么是new&#xff1f;二、new经历了什么过程&#xff1f;三、new的过程分析四、总结 JS中的new操作符 参考&#xff1a;https://www.cnblogs.com/buildnewhomeland/p/12797537.html 一、什么是new&#xff1f; 在JS中&#xff0c;new的作用是通过…
阅读更多...
【分类】分类性能评价

【分类】分类性能评价

评价指标 1、准确率、召回率、精确率、F-度量、ROC ​ 属于各类的样本的并不是均一分布&#xff0c;甚至其出现概率相差很多个数量级&#xff0c;这种分类问题称为不平衡类问题。在不平衡类问题中&#xff0c;准确率并没有多大意义&#xff0c;我们需要一些别的指标。 ​ 通…
阅读更多...
PYTHON知识点学习-函数(下)

PYTHON知识点学习-函数(下)

&#x1f308;write in front&#x1f308; &#x1f9f8;大家好&#xff0c;我是Aileen&#x1f9f8;.希望你看完之后&#xff0c;能对你有所帮助&#xff0c;不足请指正&#xff01;共同学习交流. &#x1f194;本文由 Aileen_0v0&#x1f9f8; 原创 CSDN首发&#x1f412; 如…
阅读更多...
Mac安装brew、mysql、redis

Mac安装brew、mysql、redis

mac安装brew mac安装brewmac安装mysql并配置开机启动mac安装redis并配置开机启动 mac安装brew 第一步&#xff1a;执行. /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"第二步&#xff1a;输入开机密码 第三…
阅读更多...
element+vue table表格全部数据和已选数据联动

element+vue table表格全部数据和已选数据联动

1.组件TableChoose <template><div class"tableChooseBox"><div class"tableRow"><div class"tableCard"><div class"tableHeadTip">全部{{ labelTitle }}</div><slot name"body" …
阅读更多...
《奥本海默》热映，Sam Altman 会是下个他吗？

《奥本海默》热映，Sam Altman 会是下个他吗？

撰文&#xff1a;Nathan Gardels 来源&#xff1a;Noema 治理可能摧毁社会的技术。 图片来源&#xff1a;由无界AI生成 电影导演克里斯托弗 - 诺兰&#xff08;Christopher Nolan&#xff09;说&#xff0c;他曾与正在经历“奥本海默时刻”的人工智能科学家交谈过&#xff0c;他…
阅读更多...
JavaScript 实现树形结构和一维数组互相转换

JavaScript 实现树形结构和一维数组互相转换

背景 树形结构和一维数组是开发中很容易碰到的情况&#xff0c;也是面试中很容易碰到的手撕题目 实现 一、一维数组转树形结构 FROM const source [{ id: 1, name: "张三", pid: 0 },{ id: 2, name: "李四", pid: 1 },{ id: 3, name: "王五&qu…
阅读更多...
【【STM32--28--IO引脚的复用功能】】

【【STM32--28--IO引脚的复用功能】】

STM32–28–IO引脚的复用功能 STM32的IO复用功能 何为复用? 我们先了解一下何为通用 IO端口的输入或输出是由GPIO外设控制&#xff0c;我们称之为通用 复用&#xff1a; IO端口的输入或者是输出是由其他非GPIO外设控制就像经常说的USART 由 DR寄存器进行输出 STM32的IO复用功…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023