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

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

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

相关技术方案

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

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

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.rhkb.cn/news/122347.html

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

相关文章

leetcode172. 阶乘后的零(java)

leetcode172. 阶乘后的零(java)

阶乘后的零 题目描述巧妙的解法代码演示 上期经典 题目描述 难度 - 中等 172. 阶乘后的零 给定一个整数 n &#xff0c;返回 n! 结果中尾随零的数量。 提示 n! n * (n - 1) * (n - 2) * … * 3 * 2 * 1 示例 1&#xff1a; 输入&#xff1a;n 3 输出&#xff1a;0 解释&#…
阅读更多...
【计算机基础】Git从安装到使用，详细每一步！扩展Github\Gitlab

【计算机基础】Git从安装到使用，详细每一步！扩展Github\Gitlab

&#x1f4e2;&#xff1a;如果你也对机器人、人工智能感兴趣&#xff0c;看来我们志同道合✨ &#x1f4e2;&#xff1a;不妨浏览一下我的博客主页【https://blog.csdn.net/weixin_51244852】 &#x1f4e2;&#xff1a;文章若有幸对你有帮助&#xff0c;可点赞 &#x1f44d;…
阅读更多...
面试题速记：JavaScript有哪些数据类型，它们的区别是？

面试题速记：JavaScript有哪些数据类型，它们的区别是？

JavaScript有哪些数据类型&#xff0c;它们的区别&#xff1f; JavaScript共有八种数据类型&#xff0c;分别是 Undefined、Null、Boolean、Number、String、Object、Symbol、BigInt。 其中 Symbol 和 BigInt 是ES6 中新增的数据类型&#xff1a; ●Symbol 代表创建后独一无二…
阅读更多...
《安富莱嵌入式周报》第321期：开源12导联便携心电仪，PCB AI设计，150M示波器差分探头，谷歌全栈环境IDX，微软在Excel推出Python

《安富莱嵌入式周报》第321期：开源12导联便携心电仪，PCB AI设计，150M示波器差分探头，谷歌全栈环境IDX，微软在Excel推出Python

周报汇总地址&#xff1a;嵌入式周报 - uCOS & uCGUI & emWin & embOS & TouchGFX & ThreadX - 硬汉嵌入式论坛 - Powered by Discuz! 视频版&#xff1a; https://www.bilibili.com/video/BV1ju4y1D7A8/ 《安富莱嵌入式周报》第321期&#xff1a;开源12导…
阅读更多...
机器学习笔记之最优化理论与方法(七)无约束优化问题——常用求解方法(上)

机器学习笔记之最优化理论与方法(七)无约束优化问题——常用求解方法(上)

机器学习笔记之最优化理论与方法——基于无约束优化问题的常用求解方法[上] 引言总体介绍回顾&#xff1a;线搜索下降算法收敛速度的衡量方式线性收敛范围高阶收敛范围 二次终止性朴素算法&#xff1a;坐标轴交替下降法最速下降法(梯度下降法)梯度下降法的特点 针对最速下降法缺…
阅读更多...
《vue3实战》运用push（）方法实现电影评价系统的添加功能

《vue3实战》运用push（）方法实现电影评价系统的添加功能

目录 前言 电影评价系统的添加功能是什么&#xff1f; 电影评价系统的添加功能有什么作用&#xff1f; 一、push&#xff08;&#xff09;方法是什么&#xff1f;它有什么作用&#xff1f; 含义&#xff1a; 作用&#xff1a; 二、功能实现 这段是添加开始时点击按钮使…
阅读更多...
用户端APP自动化测试_L2

用户端APP自动化测试_L2

目录&#xff1a; appium server 环境安装capability 进阶用法元素定位工具高级定位技巧-xpath 定位高级定位技巧-css 定位与原生定位特殊控件 toast 识别显式等待高级使用高级控件交互方法设备交互api模拟器控制雪球财经app股票详情功能点自动化测试实战 1.appium server 环…
阅读更多...
Podman安装与使用

Podman安装与使用

1.Podman简介 Podman是一个无守护进程的容器引擎&#xff0c;用于在Linux系统上开发、管理和运行OCI容器。 Podman的主要功能包括&#xff1a; 创建和管理容器&#xff1a;Podman可以创建、启动、停止和删除容器&#xff0c;以及管理容器的生命周期。容器镜像管理&#xff1…
阅读更多...
WSL中为Ubuntu和Debian设置固定IP的终极指南

WSL中为Ubuntu和Debian设置固定IP的终极指南

文章目录 **WSL中为Ubuntu和Debian设置固定IP的终极指南****引言/背景****1. 传统方法****2. 新方法：添加指定IP而不是更改IP****结论**WSL中为Ubuntu和Debian设置固定IP的终极指南 引言/背景 随着WSL（Windows Subsystem for Linux）的普及，越来越多的开发者开始在Windows…
阅读更多...
77 # koa 中间件的应用

77 # koa 中间件的应用

调用 next() 表示执行下一个中间件 const Koa require("koa");const app new Koa();app.use(async (ctx, next) > {console.log(1);next();console.log(2); });app.use(async (ctx, next) > {console.log(3);next();console.log(4); });app.use(async (ctx,…
阅读更多...
说说 TCP的粘包、拆包

说说 TCP的粘包、拆包

分析&回答 拆包和粘包是在socket编程中经常出现的情况&#xff0c; 在socket通讯过程中&#xff0c;如果通讯的一端一次性连续发送多条数据包&#xff0c;tcp协议会将多个数据包打包成一个tcp报文发送出去&#xff0c;这就是所谓的粘包。如果通讯的一端发送的数据包超过一…
阅读更多...
MRI多任务技术及应用

MRI多任务技术及应用

目录 一、定量心血管磁共振成像&#xff08;CMR&#xff09;的改进方法二、磁共振多任务三、磁共振多任务的成像框架四、磁共振多任务的图像模型和采样和重建策略五、利用MR多任务进行快速三维稳态CEST(ss-CEST)成像5.1 利用MR多任务进行快速三维稳态CEST(ss-CEST)成像介绍5.2 …
阅读更多...
【数据结构】链表

【数据结构】链表

【数据结构】 链表 1.链表的概念及结构 链表是一种物理存储单元上非连续、非顺序的存储结构&#xff0c;数据元素的逻辑顺序是通过链表中的指针链接次序实现的。链表由一系列结点&#xff08;链表中每一个元素称为结点&#xff09;组成&#xff0c;结点可以在运行时动态生成。…
阅读更多...
C# void 关键字学习

C# void 关键字学习

C#中void关键字是System.Void的别名&#xff1b; 可以将 void 用作方法&#xff08;或本地函数&#xff09;的返回类型来指定该方法不返回值&#xff1b; 如果C&#xff03;方法中没有参数&#xff0c;则不能将void用作参数&#xff1b;这是与C语言不同的&#xff0c;C语言有…
阅读更多...
rk3568 SDK的buildroot添加package

rk3568 SDK的buildroot添加package

开发源码工程 首先进入<SDK>/app 目录下&#xff0c;在该目录下创建一个名为“mypackage”的文件夹。 在 mypackage 目录下创建一个.c 源文件 main.c&#xff0c;以及一个 Makefile 文件。 大家可以自己在 main.c 源文件中编写一个简单的测试代码&#xff0c;譬如打印一…
阅读更多...
设计模式-建造者（生成器）模式

设计模式-建造者（生成器）模式

文章目录 简介建造者模式的核心概念产品&#xff08;Product&#xff09;建造者&#xff08;Builder&#xff09;指挥者&#xff08;Director&#xff09;建造者模式与其他设计模式的关系工厂模式和建造者模式uml对比 建造者模式的实现步骤建造者模式的应用场景spring中应用 建…
阅读更多...
RabbtiMQ的安装与使用

RabbtiMQ的安装与使用

一、安装Erlang与Rabbitmq 安装教程本教程是在centos8下试验的&#xff0c;其实linux系统的都差不多RabbitMQ官方&#xff1a;Messaging that just works — RabbitMQRabbitMQ是开源AMQP实现&#xff0c;服务器端用Erlang语言编写&#xff0c;Python、Ruby、 NET、Java、JMS、c…
阅读更多...
软件架构之前后端分离架构服务器端高并发演进之路

软件架构之前后端分离架构服务器端高并发演进之路

软件架构之前后端分离架构&服务器端高并发演进之路 前后端分离架构从业务角度从质量属性从性能角度 服务器端关于不同并发量的演进之路1. 单体架构2. 第一次演进&#xff1a;应用服务器和数据库服务器分开部署3. 第二次演进&#xff1a;引入本地缓存和分部署缓存4. 第三次演…
阅读更多...
录屏没有声音？录制声音，3招教你搞定

录屏没有声音？录制声音，3招教你搞定

在录制屏幕内容时&#xff0c;声音是不可或缺的要素之一&#xff0c;可以有效地增强录制视频的表现力和传达效果。然而&#xff0c;有时候可能会遇到录屏没有声音的情况&#xff0c;这可能会让录制的视频失去一部分重要信息。本文将为您介绍录屏录声音的3种方法&#xff0c;帮助…
阅读更多...
nios里面打开eclipse遇到Unresolved inclusion: “system.h“等问题

nios里面打开eclipse遇到Unresolved inclusion: “system.h“等问题

问题&#xff1a;在Nios中打开软核部分代码时&#xff0c;遇到一堆Unresolved inclusion: "system.h"等问题报错 原因&#xff1a;bsp文件和软核没关联&#xff0c;导致找不到头文件地址&#xff0c;关联一下就好 解决步骤&#xff1a; 右键bsp文件&#xff0c;点击…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023