springboot框架使用RabbitMQ举例代码
以前分享过一个理论有兴趣的小伙伴可以看下
https://blog.csdn.net/Drug_/article/details/138164180
不多说 还是直接上代码
第一步:引入依赖 可以不指定版本
<!-- amqp --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
第二步 配置文件
#配置rabbitMq 服务器rabbitmq:host: ${rabbitmq.rabbitmqHost}port: ${rabbitmq.rabbitmqPort}username: ${rabbitmq.rabbitmqUsername}password: ${rabbitmq.rabbitmqPassword}virtual-host: ${rabbitmq.rabbitmqVhost}connection-timeout: 5000#消费者配置listener:simple:
# 不建议使用 自带的重试配置 因为有几种情况会失效 在网上摘抄的 网友的测试
# 重试机制使用场景:
# 1. 如果是业务代码,比如空指针之类的异常那重试机制其实没什么用
# 2. 代码中不能使用try/catch捕获异常,否则重试机制失效
# 我在消费者 使用了 try 发现 确实失效了 所以 我觉得 需要手动在消费者里累计重试次数 自行处理异常
# retry:
# enabled: true #开启消费者retry重试机制
# max-attempts: 3 # 最大重试次数
# multiplier: 2.0 # 重试间隔时间倍数
# initial-interval: 1000 # 初始重试间隔时间(毫秒)
# max-interval: 10000 # 最大重试间隔时间(毫秒)acknowledge-mode: manual # 手动确认消息,防止消息丢失 auto manual手动确认模式default-requeue-rejected: true #是否将拒绝的消息重新入队。默认是 true,即拒绝的消息会重新入队。 配合手动确认模式concurrency: 1 #: 消费者线程池的并发数。设置同时处理的消费者数量max-concurrency: 1 #最大并发消费者数量prefetch: 1 # 限制每个消费者一次可以获取的消息数量,防止消息在某个消费者身上发生阻塞#生产者配置
# publisher-returns: true # 启用发布者返回模式。设置为 true 启用,确保如果消息无法路由到目标队列,则会返回给生产者。# none: 不启用发布者确认。# correlated: 启用发布者确认并使用 CorrelationData 对象,可以在回调中进行处理。#: 启用简单的发布者确认模式,不带 CorrelationData。
# publisher-confirm-type: none
第三步定义常量 :
package com.testweb.testweb.rabbitmq.web;/*** User:Json* Date: 2024/9/3**/
public class MqConstant {public static final String TestDirectRouting = "rabbitmq.TestDirectRouting";public static final String TestDirectQueue = "rabbitmq.TestDirectQueue";public static final String TestDirectExchange = "rabbitmq.TestDirectExchange";
}
第四步 消费者定义:
package com.testweb.testweb.rabbitmq.web.consumer;import com.testweb.testweb.rabbitmq.web.MqConstant;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.Header;
import com.rabbitmq.client.Channel;import java.util.HashMap;
import java.util.Map;/*** User:Json* Date: 2024/9/3* 消费者**/
@Configuration
public class TestConsumer {//队列@Beanpublic Queue TestDirectQueue() {Map<String, Object> args = new HashMap<>();args.put("x-ha-policy", "all"); //将队列设置为在集群中的所有节点上都可用return new Queue(MqConstant.TestDirectQueue, true, false, false, args);}@Beanpublic DirectExchange TestDirectExchange() {return new DirectExchange(MqConstant.TestDirectExchange, true, false);}@Beanpublic Binding bindingDirect() {return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(MqConstant.TestDirectRouting);}@RabbitListener(queues = MqConstant.TestDirectQueue)public void process1(Message testMessage, Channel channel) {// 消息的唯一标识idlong deliveryTag = testMessage.getMessageProperties().getDeliveryTag();//重试次数Integer retryCount =(Integer) testMessage.getMessageProperties().getHeaders().getOrDefault("retryCount", 0);try {// 处理消息的业务逻辑System.out.println("Received order message: " + new String(testMessage.getBody()));//假装异常int a= 1/0;// 手动确认消息// deliveryTag 唯一标识// multiple 是否批量拒绝,true 表示拒绝当前及之前的所有消息,false 表示仅拒绝当前消息channel.basicAck(deliveryTag, false);} catch (Exception e) {if (retryCount < 3) { // 设置最大重试次数try {System.out.println("处理失败,拒绝消息并重新入队 :" + testMessage);MessageProperties messageProperties = new MessageProperties();messageProperties.setHeader("retryCount", retryCount + 1);Message newMessage = new Message(testMessage.getBody(), messageProperties);// 重新入队 未写完channel.basicPublish(MqConstant.TestDirectExchange, MqConstant.TestDirectRouting, null, newMessage.getBody());// 手动确认原消息,防止死循环channel.basicAck(deliveryTag, false);// 处理失败,拒绝消息并重新入队 方式1// 消息标识 deliveryTag,// multiple 是否批量拒绝,true 表示拒绝当前及之前的所有消息,false 表示仅拒绝当前消息,,// requeue 是否将消息重新放回队列,true 表示重新放入队列,false 表示丢弃// channel.basicNack(deliveryTag, false, true);// 处理失败,拒绝消息并重新入队 方式2// 消息标识 deliveryTag// requeue 是否将消息重新放回队列,true 表示重新放入队列,false 表示丢弃。//channel.basicReject(long deliveryTag, boolean requeue);
// 3. 使用场景
// basicNack:
// 当你需要拒绝一批消息时,使用 basicNack 是更好的选择,尤其是当你想在消费失败时批量拒绝多条消息。
// 适用于更复杂的场景,比如一次性处理多个未确认的消息。
// basicReject:
// 当你只想拒绝当前消息时,basicReject 是一个简化的选择。它通常用于更简单的场景,只需处理当前消息即可。
// 适合处理单个消息的拒绝。//如果你在消费者 里 只写了 消息确定 没有写 如果异常后 的处理 默认是不会把消息重新放回队列的} catch (Exception nackException) {System.out.println("重新入队失败!!!");// 处理 nack 失败的情况nackException.printStackTrace();}} else {System.out.println("达到最大重试次数 将消息发送到死信队列或进行其他处理!!!");try {channel.basicReject(deliveryTag, false); // 丢弃消息或转发到死信队列} catch (Exception rejectException) {rejectException.printStackTrace();}}}}}
第五步:生产者
package com.testweb.testweb.rabbitmq.web.producer;import com.testweb.testweb.rabbitmq.web.MqConstant;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;/*** User:Json* Date: 2024/9/3* 生产者**/
@Component
public class TestProducer {@ResourceAmqpTemplate amqpTemplate;public <T> void produce(T payload){amqpTemplate.convertAndSend(MqConstant.TestDirectExchange,MqConstant.TestDirectRouting, payload);}
}
第六步 测试:
package com.testweb.testweb.rabbitmq.web.controller;import com.testweb.testweb.rabbitmq.web.MqConstant;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;/*** User:Json* Date: 2024/9/3**/
@RestController
@RequestMapping("/testMq")
public class TestMqController {@ResourceAmqpTemplate amqpTemplate;@GetMapping("test")@CrossOrigin(origins = "*")public void test(@RequestParam String msg){amqpTemplate.convertAndSend(MqConstant.TestDirectExchange,MqConstant.TestDirectRouting, msg);}}相关文章:
springboot框架使用RabbitMQ举例代码
以前分享过一个理论有兴趣的小伙伴可以看下 https://blog.csdn.net/Drug_/article/details/138164180 不多说 还是直接上代码 第一步:引入依赖 可以不指定版本 <!-- amqp --><dependency><groupId>org.springframework.boot</groupId…...
Java实现一个延时队列
文章目录 前言正文一、基本概念1.1 延时队列的特点1.2 常见的实现方式 二、Java原生的内存型延时队列2.1 定义延时元素DelayedElement2.2 定义延时队列管理器DelayedQueueManager2.3 消费元素2.4 调试2.5 调试结果2.6 精髓之 DelayQueue.poll() 三、基于Redisson的延时队列3.1 …...
为什么说vue是双向数据流
Vue.js 被称为 双向数据绑定(two-way data binding),是因为它支持数据在 视图(View) 和 模型(Model) 之间双向流动。这意味着,当 数据变化 时,视图会自动更新;…...
创造属于你的 Claude Prompt 和个性化 SVG 卡片|对李继刚老师提示词的浅浅解析与总结
❤️ 如果你也关注大模型与 AI 的发展现状,且对大模型应用开发非常感兴趣,我会快速跟你分享最新的感兴趣的 AI 应用和热点信息,也会不定期分享自己的想法和开源实例,欢迎关注我哦! 🥦 微信公众号ÿ…...
redis与本地缓存
本地缓存是将数据存储在应用程序所在的本地内存中的缓存方式。既然,已经有了 Redis 可以实现分布式缓存了,为什么还需要本地缓存呢?接下来,我们一起来看。 为什么需要本地缓存? 尽管已经有 Redis 缓存了,但…...
git撤销commit和add
撤销commit git reset --soft HEAD^撤销add git reset .查看状态 git status...
【361】基于springboot的招生宣传管理系统
摘 要 使用旧方法对招生宣传管理系统的信息进行系统化管理已经不再让人们信赖了,把现在的网络信息技术运用在招生宣传管理系统的管理上面可以解决许多信息管理上面的难题,比如处理数据时间很长,数据存在错误不能及时纠正等问题。这次开发的招…...
【一些关于Python的信息和帮助】
Python是一种广泛使用的高级编程语言,它的设计哲学强调代码的可读性和简洁的语法(尤其是使用空格缩进划分代码块,而不是使用大括号或关键字)。Python支持多种编程范式,包括面向对象、命令式、函数式和过程式编程。 以…...
creo toolkit二次开发学习之程序集(ProAsmcomp)和装配体组件路径对象(ProAsmcomppath)
程序集ProAsmcomp可以理解为装配体组件对象。 对象ProAssembly是ProSolid的一个实例,并共享相同的声明。因此,ProAssembly对象可以作为适用于装配体的任何ProSolid和ProMdl函数的输入。特别是,因为你可以使用函数ProSolidFeatVisit()来遍历特…...
深入浅出 Spring Boot 与 Shiro:构建安全认证与权限管理框架
一、Shiro框架概念 (一)Shiro框架概念 1.概念: Shiro是apache旗下一个开源安全框架,它对软件系统中的安全认证相关功能进行了封装,实现了用户身份认证,权限授权、加密、会话管理等功能,组成一…...
外包干了三年,精神严重内耗...
前段时间我同事(做测试的一个妹子)跟我讲,感觉早上起来十分的疲惫,不想上班,问我们这是什么样的现象,其实有时候我也有这种感觉,虽然我卷,但我也是肉体凡胎啊!不是机器人…...
ruoyi-vue集成tianai-captcha验证码
后端代码 官方使用demo文档:http://doc.captcha.tianai.cloud/#%E4%BD%BF%E7%94%A8demo 我的完整代码:https://gitee.com/Min-Duck/RuoYi-Vue.git 主pom.xml 加入依赖 <!-- 滑块验证码 --><dependency><groupId>cloud.tianai.captc…...
Django安装
在终端创建django项目 1.查看自己的python版本 输入对应自己本机python的版本,列如我的是3.11.8 先再全局安装django依赖包 2.在控制窗口输入安装命令: pip3.11 install django 看到Successflully 说明我们就安装成功了 python的Scripts文件用于存…...
Ubuntu 20.04 安装 QGC v4.3 开发环境
Ubuntu 20.04 安装 QGC开发环境 1. 准备安装 Qt 5.15.2安装依赖获取源码 2. 编译参考 前言 QGC ( QGroundControl) 是一个开源地面站,基于QT开发的,有跨平台的功能。可以在Windows,Android,MacOS或Linux上运行。它可以将PX4固件加…...
WPF+MVVM案例实战(二十一)- 制作一个侧边弹窗栏(AB类)
文章目录 1、案例效果1、侧边栏分类2、AB类侧边弹窗实现1.文件创建2、样式代码与功能代码实现3、功能代码实现 3 运行效果4、源代码获取 1、案例效果 1、侧边栏分类 A类 :左侧弹出侧边栏B类 :右侧弹出侧边栏C类 :顶部弹出侧边栏D类 …...
linux中怎样登录mysql数据库
在Linux中登录MySQL数据库,可以使用以下命令: mysql -u username -p 其中,username是你的MySQL用户名。运行该命令后,系统会提示你输入密码。 如果MySQL服务器不在本地主机或者你需要指定不同的端口,可以使用以下命…...
深入理解 Linux 内存管理:free 命令详解
在 Linux 系统中,内存是关键的资源之一,管理和监控内存的使用情况对系统的稳定性和性能至关重要。free 命令是 Linux 中用于查看内存使用情况的重要工具,它可以让我们快速了解系统中物理内存和交换分区(Swap)的使用状态…...
指针万字超级最强i解析与总结!!!!!
文章目录 1.内存和地址1.1内存1.2究竟该如何理解编址 2.指针变量和地址2.1 取地址操作符(&)2.2指针变量和解引用操作符(*)2.2.1指针变量2.2.2如何拆解指针类型2.2.3解引用操作符 2.3 指针变量的大小 3.指针变量类型的意义3.1指…...
告别生硬电子音,这款TTS软件让语音转换更自然动听
Balabolka是一款革新性的文本语音转换工具,为用户提供了极其灵活和个性化的阅读体验。这款软件不仅仅是简单的文字朗读器,更是一个智能的语音助手,能够将各类文本瞬间转化为生动自然的语音输出。 软件的核心优势在于其卓越的文件兼容性和多样…...
CORS(跨域资源共享)和SOP(同源策略)
CORS(跨域资源共享)和SOP(同源策略)不是同一个东西,但它们紧密相关,并且常常一起讨论,因为 CORS 是为了解决同源策略带来的跨域问题而引入的。 同源策略(Same-Origin Policy&#x…...
使用docker在3台服务器上搭建基于redis 6.x的一主两从三台均是哨兵模式
一、环境及版本说明 如果服务器已经安装了docker,则忽略此步骤,如果没有安装,则可以按照一下方式安装: 1. 在线安装(有互联网环境): 请看我这篇文章 传送阵>> 点我查看 2. 离线安装(内网环境):请看我这篇文章 传送阵>> 点我查看 说明:假设每台服务器已…...
XCTF-web-easyupload
试了试php,php7,pht,phtml等,都没有用 尝试.user.ini 抓包修改将.user.ini修改为jpg图片 在上传一个123.jpg 用蚁剑连接,得到flag...
模型参数、模型存储精度、参数与显存
模型参数量衡量单位 M:百万(Million) B:十亿(Billion) 1 B 1000 M 1B 1000M 1B1000M 参数存储精度 模型参数是固定的,但是一个参数所表示多少字节不一定,需要看这个参数以什么…...
IGP(Interior Gateway Protocol,内部网关协议)
IGP(Interior Gateway Protocol,内部网关协议) 是一种用于在一个自治系统(AS)内部传递路由信息的路由协议,主要用于在一个组织或机构的内部网络中决定数据包的最佳路径。与用于自治系统之间通信的 EGP&…...
家政维修平台实战20:权限设计
目录 1 获取工人信息2 搭建工人入口3 权限判断总结 目前我们已经搭建好了基础的用户体系,主要是分成几个表,用户表我们是记录用户的基础信息,包括手机、昵称、头像。而工人和员工各有各的表。那么就有一个问题,不同的角色…...
【CSS position 属性】static、relative、fixed、absolute 、sticky详细介绍,多层嵌套定位示例
文章目录 ★ position 的五种类型及基本用法 ★ 一、position 属性概述 二、position 的五种类型详解(初学者版) 1. static(默认值) 2. relative(相对定位) 3. absolute(绝对定位) 4. fixed(固定定位) 5. sticky(粘性定位) 三、定位元素的层级关系(z-i…...
UR 协作机器人「三剑客」:精密轻量担当(UR7e)、全能协作主力(UR12e)、重型任务专家(UR15)
UR协作机器人正以其卓越性能在现代制造业自动化中扮演重要角色。UR7e、UR12e和UR15通过创新技术和精准设计满足了不同行业的多样化需求。其中,UR15以其速度、精度及人工智能准备能力成为自动化领域的重要突破。UR7e和UR12e则在负载规格和市场定位上不断优化…...
Caliper 配置文件解析:config.yaml
Caliper 是一个区块链性能基准测试工具,用于评估不同区块链平台的性能。下面我将详细解释你提供的 fisco-bcos.json 文件结构,并说明它与 config.yaml 文件的关系。 fisco-bcos.json 文件解析 这个文件是针对 FISCO-BCOS 区块链网络的 Caliper 配置文件,主要包含以下几个部…...
论文笔记——相干体技术在裂缝预测中的应用研究
目录 相关地震知识补充地震数据的认识地震几何属性 相干体算法定义基本原理第一代相干体技术:基于互相关的相干体技术(Correlation)第二代相干体技术:基于相似的相干体技术(Semblance)基于多道相似的相干体…...
【SSH疑难排查】轻松解决新版OpenSSH连接旧服务器的“no matching...“系列算法协商失败问题
【SSH疑难排查】轻松解决新版OpenSSH连接旧服务器的"no matching..."系列算法协商失败问题 摘要: 近期,在使用较新版本的OpenSSH客户端连接老旧SSH服务器时,会遇到 "no matching key exchange method found", "n…...
