Spring AMQP-保证消费者消息的可靠性
为什么要保证消息的可靠性?
当MQ向消费者发送一个消息之后需要得到消费者的状态,因为消息并不一定就真的被消费者给消费了,可能在消费的过程中出现了一些意外,比如
1. 网络问题
2. 消息转换有问题
3. 消费者本身的业务处理有问题
消费者确认机制
消费者消息处理状态:
- ack:消息成功接收,并且成功被处理,MQ将此消息删除
- nack:消息处理失败,需要MQ重新发送消息
- reject:消息处理失败并且拒绝该消息,MQ将此消息删除
由于消息回执的处理代码比较统一,因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式,有三种模式:
none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用
manual:手动模式。需要自己在业务代码中调用api,发送ack或reject,存在业务入侵,但更灵活
auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack. 当业务出现异常时,根据异常判断返回不同结果:
如果是业务异常,会自动返回
nack;如果是消息处理或校验异常,自动返回
reject;
在配置文件中通过下面的配置即可设置ACK的处理方式
spring:rabbitmq:listener:simple:acknowledge-mode: c # none 默认 auto 自动确认 manual 手动确认
消费者重试机制
消费者接收了一个消息,但是在处理的过程中出现异常了,那么AMQP会不断的重试,直到把资源占完然后崩掉,这个时候就必须要设置重试机制,限制重试的次数,避免无限制重试。
spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false测试
先别配置重试机制,然后在需要在接收消息的地方手动抛出一个异常,查看控制台就会看见消费者在尝试不断的获取消息,但是一直获取不到无限制的重试
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "fanout.hamll.query1", // 队列名称durable = "true"), // 是否持久exchange = @Exchange(name = "fanout.hamll", type = ExchangeTypes.FANOUT) // 交换机名称))public void query2(String message) {System.err.println("fanout.hamll.query1 消息内容:" + message); throw new RuntimeException("故意的错误"); // 抛出异常}![]()
配置好重试之后到了三次就会直接停止,这样子就很好的减少了系统资源的消耗
业务的幂等性判断
什么是幂等性?
在Java领域,幂等性是指同一个请求,不管发送多少次执行的结构都是一样的。
比如支付和交易,支付成功之后通知交易服务修改状态。在交易服务需要查询订单并判断订单的状态,这样子不管同一个订单重复发起多少次请求,都不会对业务的结果造成影响。
MQ保证消息的幂等性
MQ中的幂等是说,不管消息是否被重复消费,都不会对业务造成影响、处理的结果都是一致的。
MQ实现业务幂等性
为每个消息都创建一个唯一的MessageId在操作的时候将其存入数据库,然后在进行判断消息是否存在,存在就直接跳过业务的处理,不存在就继续操作。
@Beanpublic MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;}
业务实现幂等性
在业务的操作中,比如支付和交易服务,支付成功之后会通知交易服务修改订单的状态,而在交易服务应该做判断,判断该订单的状态是否未未支付。如果是未支付就继续处理接下来的业务,否则就直接结束。
package com.hmall.trade.listener;import com.hmall.trade.domain.po.Order; import com.hmall.trade.service.IOrderService; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;@Component @Slf4j public class PayStatusListener {@AutowiredIOrderService orderService;@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "trade.pay.success.queue", durable = "true"),exchange = @Exchange(value = "pay.direct", type = ExchangeTypes.DIRECT),key = "pay.success"))public void paySuccess(Long orderId) {log.info("支付成功,订单号:{}", orderId);//查询当前订单 判断幂等性Order order = orderService.getById(orderId);//判断状态以及对象是否存在if (order == null || order.getStatus() != 1) {return;}orderService.markOrderPaySuccess(orderId);} }
相关文章:
Spring AMQP-保证消费者消息的可靠性
为什么要保证消息的可靠性? 当MQ向消费者发送一个消息之后需要得到消费者的状态,因为消息并不一定就真的被消费者给消费了,可能在消费的过程中出现了一些意外,比如 1. 网络问题 2. 消息转换有问题 3. 消费者本身的业务处理有问题 …...
Linux(Centos 7.6)命令详解:mkdir
1.命令作用 如果目录还不存在,则创建目录(Create the DIRECTORY, if they do not already exist.) 2.命令语法 Usage: mkdir [OPTION]... DIRECTORY... 3.参数详解 OPTION: -m, --modeMODE,创建新目录同时设置权限模式-p, --parents,创…...
在K8S上部署OceanBase的最佳实践
在K8S上部署OceanBase的最佳实践 目录 1. 背景与选型 1.1 为什么选择OB1.2 为什么选择ob-operator实现OB on K8S 2. 部署实操 2.1 环境准备2.2 安装 ob-operator2.3 配置 OB 集群2.4 配置 OBProxy 集群2.5 Headless Service 和 CoreDNS 配置2.6 监控与运维 2.6.1 Promethues部…...
IDEA中Maven依赖包导入失败报红的潜在原因
在上网试了别人的八个问题总结之后依然没有解决: IDEA中Maven依赖包导入失败报红问题总结最有效8种解决方案_idea导入依赖还是报红-CSDN博客https://blog.csdn.net/qq_43705131/article/details/106165960 江郎才尽之后突然想到一个原因:<dep…...
【计算机网络】课程 实验五 静态路由配置
实验五 静态路由配置 一、实验目的 理解静态路由的工作原理,掌握如何配置静态路由。 二、实验分析与设计 【背景描述】 假设校园网分为 2 个区域,每个区域内使用 1 台路由器连接 2 个子网, 现要在路由器上 做适当配置,实现校…...
基于单片机的数字气压计设计
摘要:在嵌入式技术快速发展过程中,智能测量仪器被广泛应用于工业生产以及人们日常生活领域。数字气压计在实际应用中,利用气压传感器检测环境中的压力大小,便于实现对设备进行智能化的控制操作。数字气压计在气象监测、矿产开采、科学实验等环…...
【Docker项目实战】使用Docker部署Typemill轻量级平面文件CMS
【Docker项目实战】使用Docker部署Typemill轻量级平面文件CMS 一、Typemill介绍1.1 Typemill简介1.2 主要特点1.3 主要使用场景二、本次实践规划2.1 本地环境规划2.2 本次实践介绍三、本地环境检查3.1 检查Docker服务状态3.2 检查Docker版本3.3 检查docker compose 版本四、下载…...
react ts 定义基本类型,组件通过ref调用时类型提示
记录,以防忘记 子组件 import React, { forwardRef, Ref, useImperativeHandle, useState } from react;// 类型定义方式1 interface IProps {/**参数1 */params1: number | string | undefined/**参数2 */params2: number | string | undefined/**方法 */openDia…...
二十三种设计模式-原型模式
原型模式(Prototype Pattern)是一种创建型设计模式,它通过拷贝现有的实例来创建新的实例,而不是通过新建实例。这种方式可以避免复杂的构造过程,同时还能保持对象的创建和使用分离,提高系统的灵活性和扩展性…...
提升汽车金融租赁系统的效率与风险管理策略探讨
内容概要 在汽车金融租赁系统这个复杂的生态中,提升整体效率是每个企业都渴望达成的目标。首先,优化业务流程是实现高效运行的基础。通过分析目前的流程,找出冗余环节并进行简化,能够帮助企业缩短审批时间,提高客户满…...
Spring Framework 5.3.x源码构建 (jdk-1.8, gradle 7.5.1, idea2024.3)
1、下载jdk安装并配置环境变量(自行百度) https://www.oracle.com/java/technologies/downloads/#java8 2、下载spring-framework源码,切换分支到5.3.x https://github.com/spring-projects/spring-framework.git 备用地址 https://gitco…...
leetcode 2241. 设计一个 ATM 机器 中等
一个 ATM 机器,存有 5 种面值的钞票:20 ,50 ,100 ,200 和 500 美元。初始时,ATM 机是空的。用户可以用它存或者取任意数目的钱。 取款时,机器会优先取 较大 数额的钱。 比方说,你想…...
IO模型与NIO基础
File类 File类主要是JAVA为文件这块的操作(如删除、新建等)而设计的相关类File类的包名是java.io,其实现了Serializable, Comparable两大接口以便于其对象可序列化和比较 创建一个文件/文件夹 删除文件/文件夹 获取文件/文件夹 判断文件/文件夹是否存在 对文件夹进…...
上门按摩系统架构与功能分析
一、系统架构 服务端:Java(最低JDK1.8,支持JDK11以及JDK17)数据库:MySQL数据库(标配5.7版本,支持MySQL8)ORM框架:Mybatis(集成通用tk-mapper,支持…...
ubuntu安装ssh9.2
删除旧版本: dpkg --list|grep ssh apt remove sshwget https://ftp.openbsd.org/pub/OpenBSD/OpenSSH/portable/openssh-9.2p1.tar.gz tar xzvf openssh-9.2p1.tar.gz cd openssh-9.2p1 #下载依赖#开始编译安装 ./configure && make && make inst…...
linux wsl配置 redis远程连接
✅ 1. 修改 Redis 配置文件 在 WSL 的 Redis 配置文件中,找到 redis.conf 或 /etc/redis/redis.conf 文件,编辑以下配置项: ➡️ 更新 bind 配置项 将 bind 127.0.0.1 ::1 修改为: bind 0.0.0.0这样,Redis 将监听所…...
JVM 优化指南
JVM 优化指南 1. JVM 参数配置 1.1 基础参数配置 设置堆内存大小 -Xms2048m -Xmx2048m 设置新生代大小 -Xmn1024m 设置元空间大小 -XX:MetaspaceSize256m -XX:MaxMetaspaceSize256m 设置线程栈大小 -Xss512k1.2 垃圾回收器配置 使用 G1 垃圾回收器 -XX:UseG1GC 设置期望停顿…...
关机重启后,GitLab服务异常
整理机房,关闭了所有主机重新上架。 上架后开机,所有主机硬件启动正常。 其中一台GitLab服务器启动正常,使用gitlab-ctl status查看服务业正常。 但使用web登陆却失败,如下图: 反复测试,发现无论使用正确密码还是错误密码都是同样的提示。很大可能是数据库的问题。 使…...
谷粒商城-高级篇完结-Sleuth+Zipkin 服务链路追踪
1、基本概念和整合 1.1、为什么用 微服务架构是一个分布式架构,它按业务划分服务单元,一个分布式系统往往有很多个服务单元。由于服务单元数量众多,业务的复杂性,如果出现了错误和异常,很难去定位 。主要体现在&#…...
C语言基本知识复习浓缩版:标识符、函数、进制、数据类型
C语言基本知识复习浓缩版:标识符、函数、进制、数据类型 【c语言期末复习3小时速成【完整全集】期末速成含考试题c语言期末速成突击复习C语言补考C语言期末大一】 B站看到的复习C语言视频,感觉非常棒,就跟着进行了一下学习。众所周知&#…...
OpenLayers 可视化之热力图
注:当前使用的是 ol 5.3.0 版本,天地图使用的key请到天地图官网申请,并替换为自己的key 热力图(Heatmap)又叫热点图,是一种通过特殊高亮显示事物密度分布、变化趋势的数据可视化技术。采用颜色的深浅来显示…...
地震勘探——干扰波识别、井中地震时距曲线特点
目录 干扰波识别反射波地震勘探的干扰波 井中地震时距曲线特点 干扰波识别 有效波:可以用来解决所提出的地质任务的波;干扰波:所有妨碍辨认、追踪有效波的其他波。 地震勘探中,有效波和干扰波是相对的。例如,在反射波…...
java_网络服务相关_gateway_nacos_feign区别联系
1. spring-cloud-starter-gateway 作用:作为微服务架构的网关,统一入口,处理所有外部请求。 核心能力: 路由转发(基于路径、服务名等)过滤器(鉴权、限流、日志、Header 处理)支持负…...
有限自动机到正规文法转换器v1.0
1 项目简介 这是一个功能强大的有限自动机(Finite Automaton, FA)到正规文法(Regular Grammar)转换器,它配备了一个直观且完整的图形用户界面,使用户能够轻松地进行操作和观察。该程序基于编译原理中的经典…...
使用 SymPy 进行向量和矩阵的高级操作
在科学计算和工程领域,向量和矩阵操作是解决问题的核心技能之一。Python 的 SymPy 库提供了强大的符号计算功能,能够高效地处理向量和矩阵的各种操作。本文将深入探讨如何使用 SymPy 进行向量和矩阵的创建、合并以及维度拓展等操作,并通过具体…...
AI,如何重构理解、匹配与决策?
AI 时代,我们如何理解消费? 作者|王彬 封面|Unplash 人们通过信息理解世界。 曾几何时,PC 与移动互联网重塑了人们的购物路径:信息变得唾手可得,商品决策变得高度依赖内容。 但 AI 时代的来…...
佰力博科技与您探讨热释电测量的几种方法
热释电的测量主要涉及热释电系数的测定,这是表征热释电材料性能的重要参数。热释电系数的测量方法主要包括静态法、动态法和积分电荷法。其中,积分电荷法最为常用,其原理是通过测量在电容器上积累的热释电电荷,从而确定热释电系数…...
Go 并发编程基础:通道(Channel)的使用
在 Go 中,Channel 是 Goroutine 之间通信的核心机制。它提供了一个线程安全的通信方式,用于在多个 Goroutine 之间传递数据,从而实现高效的并发编程。 本章将介绍 Channel 的基本概念、用法、缓冲、关闭机制以及 select 的使用。 一、Channel…...
vulnyx Blogger writeup
信息收集 arp-scan nmap 获取userFlag 上web看看 一个默认的页面,gobuster扫一下目录 可以看到扫出的目录中得到了一个有价值的目录/wordpress,说明目标所使用的cms是wordpress,访问http://192.168.43.213/wordpress/然后查看源码能看到 这…...
day36-多路IO复用
一、基本概念 (服务器多客户端模型) 定义:单线程或单进程同时监测若干个文件描述符是否可以执行IO操作的能力 作用:应用程序通常需要处理来自多条事件流中的事件,比如我现在用的电脑,需要同时处理键盘鼠标…...
