SpringBoot 项目中使用 spring-boot-starter-amqp 依赖实现 RabbitMQ

文章目录

    • 前言
    • 1、application.yml
    • 2、RabbitMqConfig
    • 3、MqMessage
    • 4、MqMessageItem
    • 5、DirectMode
    • 6、StateConsumer:消费者
    • 7、InfoConsumer:消费者

前言

本文是工作之余的随手记,记录在工作期间使用 RabbitMQ 的笔记。

1、application.yml

  • 使用 use 属性,方便随时打开和关闭使用 MQ ,并且可以做到细化控制。
spring:rabbitmq:use: truehost: 10.100.10.100port: 5672username: wenpassword: 123456exchangeSubPush: 'exWen'queueSubPush: 'ha.queue.SubPush'routeSubPush: '1000'exchangeState: sync.ex.StatequeueState: ha.q.ServerqueueStateSync: ha.q.StateServerrouteState: stateexchangeOnlineMonitor: 'sync.ex.State'routeOnlineMonitor: 'state'queueOnlineMonitor: 'ha.q.Online'
  • pom.xml 文件中使用的是 SpringBoot 项目,使用 spring-boot-starter-amqp 依赖。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.wen</groupId><artifactId>springboot-mybatis</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><artifactId>spring-boot-starter-parent</artifactId><groupId>org.springframework.boot</groupId><version>2.5.3</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-tomcat</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.18</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.83</version></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.1</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.18</version></dependency></dependencies>
</project>

2、RabbitMqConfig

  • 配置类,将可配置的参数使用 @Value 做好配置,与 application.yml 相互对应。
package com.wen.mq;import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;@Slf4j
@Configuration
@Data
public class RabbitMqConfig {@Value("${spring.rabbitmq.use:true}")private boolean use;@Value("${spring.rabbitmq.host}")private String host;@Value("${spring.rabbitmq.port}")private int port;@Value("${spring.rabbitmq.username}")private String username;@Value("${spring.rabbitmq.password}")private String password;@Value("${spring.rabbitmq.virtual-host:}")private String virtualHost;@Value("${spring.rabbitmq.exchangeState}")private String exchangeState;@Value("${spring.rabbitmq.queueState}")private String queueState;@Value("${spring.rabbitmq.routeState}")private String routeState;@Value(("${spring.rabbitmq.queueStateSync}"))private String queueStateSync;@Value("${spring.rabbitmq.exchangeOnlineInfo}")private String exchangeOnlineInfo;@Value("${spring.rabbitmq.routeOnlineInfo}")private String routeOnlineInfo;@Value("${spring.rabbitmq.queueOnlineInfo}")private String queueOnlineInfo;@PostConstructprivate void init() {}
}

3、MqMessage

  • MQ 消息实体类
package com.wen.mq;import lombok.Data;@Data
public class MqMessage<T> {private String msgType;private String msgOrigin;private long time;private T data;}

4、MqMessageItem

  • MQ 消息实体类
package com.wen.mq;import lombok.Data;@Data
public class MqMessageItem {private long userId;private String userName;private int userAge;private String userSex;private String userPhone;private String op;}

5、DirectMode

  • 配置中心:使用 SimpleMessageListenerContainer 进行配置。
  • 新加一个消费者队列就要在这里进行配置。
package com.wen.mq;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class DirectMode {@AutowiredRabbitMqConfig rabbitMqConfig;@Autowiredprivate CachingConnectionFactory connectionFactory;@Autowiredprivate StateConsumer stateConsumer;@Autowiredprivate InfoConsumer infoConsumer;@Beanpublic SimpleMessageListenerContainer initMQ() {if (!rabbitMqConfig.isUse()) {return null;}log.info("begin!");SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);container.setConcurrentConsumers(1);container.setMaxConcurrentConsumers(1);container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认,这里改为手动确认// 设置一个队列container.setQueueNames(rabbitMqConfig.getQueueStateSync());//如果同时设置多个队列如下: 前提是队列都是必须已经创建存在的//container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3”);//另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues//container.setQueues(new Queue("TestDirectQueue",true));//container.addQueues(new Queue("TestDirectQueue2",true));//container.addQueues(new Queue("TestDirectQueue3",true));container.setMessageListener(stateConsumer);log.info("end");return container;}@Beanpublic SimpleMessageListenerContainer contactSyncContainer() {if (!rabbitMqConfig.isUse()) {return null;}log.info("contact begin");SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);container.setConcurrentConsumers(1);container.setMaxConcurrentConsumers(1);container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认,这里改为手动确认消息//设置一个队列container.setQueueNames(rabbitMqConfig.getQueueOnlineInfo());container.setMessageListener(infoConsumer);log.info("contact end");return container;}@Beanpublic Queue queueState() {if (!rabbitMqConfig.isUse()) {return null;}return new Queue(rabbitMqConfig.getQueueState());}@Beanpublic Queue queueStateSync() {if (!rabbitMqConfig.isUse()) {return null;}return new Queue(rabbitMqConfig.getQueueStateSync());}@BeanDirectExchange exchangeState() {if (!rabbitMqConfig.isUse()) {return null;}return new DirectExchange(rabbitMqConfig.getExchangeState());}@BeanBinding bindingState() {if (!rabbitMqConfig.isUse()) {return null;}return BindingBuilder.bind(queueState()).to(exchangeState()).with(rabbitMqConfig.getRouteState());}@BeanBinding bindingStateSync() {if (!rabbitMqConfig.isUse()) {return null;}return BindingBuilder.bind(queueStateSync()).to(exchangeState()).with(rabbitMqConfig.getRouteState());}// 新加一个消费者@Beanpublic Queue queueOnlineMonitor() {if (!rabbitMqConfig.isUse()) {return null;}return new Queue(rabbitMqConfig.getQueueOnlineInfo());}@BeanDirectExchange exchangeOnlineMonitor() {if (!rabbitMqConfig.isUse()) {return null;}return new DirectExchange(rabbitMqConfig.getExchangeOnlineInfo());}@BeanBinding bindingExchangeOnlineMonitor() {if (!rabbitMqConfig.isUse()) {return null;}return BindingBuilder.bind(queueOnlineMonitor()).to(exchangeOnlineMonitor()).with(rabbitMqConfig.getRouteOnlineInfo());}
}

6、StateConsumer:消费者

  • 实现 ChannelAwareMessageListener 接口,可以在这里面做相应的操作,例如存缓存,存库等。
package com.wen.mq;import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;@Slf4j
@Component
public class StateConsumer implements ChannelAwareMessageListener {@AutowiredRabbitMqConfig rabbitMqConfig;@Overridepublic void onMessage(Message message, Channel channel) throws Exception {String queueName = message.getMessageProperties().getConsumerQueue();long deliveryTag = message.getMessageProperties().getDeliveryTag();if (!rabbitMqConfig.getQueueStateSync().equals(queueName)) {String bodyStr = new String(message.getBody(), StandardCharsets.UTF_8);try {MqMessage<List<MqMessageItem>> mqMessage = JSON.parseObject(bodyStr, new TypeReference<MqMessage<List<MqMessageItem>>>() {});// 这里可以对消息做其他处理,例如存储到缓存中List<MqMessageItem> items = mqMessage.getData();if (CollectionUtil.isNotEmpty(items)) {applyToRedis(mqMessage);}log.info("consume mq msg ok, queue:{}, deliveryTag:{}, msg:{}", queueName, deliveryTag, mqMessage);channel.basicAck(deliveryTag, false);} catch (JSONException e) {log.error("parse mq msg exception, queue:{}, deliveryTag:{}", queueName, deliveryTag, e);channel.basicReject(deliveryTag, false);} catch (Exception e) {log.error("consume mq msg exception, queue:{}, deliveryTag:{}", queueName, deliveryTag, e);channel.basicReject(deliveryTag, true); //为true会重新放回队列}}}public static final String MQ_STATE_OP_REMOVE_STATE = "REMOVE_STATE";public static final String MQ_STATE_OP_CHANGE_STATE = "CHANGE_STATE";private void applyToRedis(MqMessage<List<MqMessageItem>> mqMessage) {List<MqMessageItem> data = mqMessage.getData();Map<String, List<MqMessageItem>> itemGroupByOp = data.stream().collect(Collectors.groupingBy(item -> item.getOp()));List<MqMessageItem> stateToRemove = itemGroupByOp.get(MQ_STATE_OP_REMOVE_STATE);List<MqMessageItem> stateToChange = itemGroupByOp.get(MQ_STATE_OP_CHANGE_STATE);if (CollectionUtil.isNotEmpty(stateToRemove)) {Map<Long, Set<String>> map = new HashMap<>();for (MqMessageItem item : stateToRemove) {map.computeIfAbsent(item.getUserId(), u -> new HashSet<>()).add(String.valueOf(item.getUserAge()));}// cacheService.removeUserState(map);}if (CollectionUtil.isNotEmpty(stateToChange)) {List<MqMessageItem> list = stateToChange.stream().map(u -> {MqMessageItem dto = new MqMessageItem();dto.setUserId(u.getUserId());dto.setUserAge(u.getUserAge());dto.setUserName(u.getUserName());dto.setUserSex(u.getUserSex());dto.setUserPhone(u.getUserPhone());return dto;}).collect(Collectors.toList());// cacheService.saveUserState(list);}}
}

7、InfoConsumer:消费者

  • 实现 ChannelAwareMessageListener 接口,可以在这里面做相应的操作,例如存缓存,存库等。
package com.wen.mq;import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class InfoConsumer implements ChannelAwareMessageListener {@AutowiredRabbitMqConfig rabbitMqConfig;@Overridepublic void onMessage(Message message, Channel channel) throws Exception {String queueName = message.getMessageProperties().getConsumerQueue();log.info("queueName: {}", queueName);long deliveryTag = message.getMessageProperties().getDeliveryTag();try {byte[] body = message.getBody();String content = new String(body);MqMessage msg = JSONObject.parseObject(content, MqMessage.class);if (rabbitMqConfig.getQueueOnlineInfo().equals(queueName)) {// 订阅到的消息就是变更的消息// 这里可使用service对消息进行消费,返回一个booleanlog.info("用户监控数据写入失败!数据:{}", msg);}log.info("consume mq msg ok, queue:{}, deliveryTag:{}, msg:{}", queueName, deliveryTag, msg);channel.basicAck(deliveryTag, false);} catch (JSONException e) {log.error("parse mq msg exception, queue:{}, deliveryTag:{}", queueName, deliveryTag, e);channel.basicReject(deliveryTag, false); //为true会重新放回队列} catch (Exception e) {log.error("consume mq msg exception, queue:{}, deliveryTag:{}", queueName, deliveryTag, e);channel.basicReject(deliveryTag, true); //为true会重新放回队列}}
}

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

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

相关文章

【优选算法篇】两队接力跑:双指针协作解题的艺术(下篇)

文章目录 须知 &#x1f4ac; 欢迎讨论&#xff1a;如果你在学习过程中有任何问题或想法&#xff0c;欢迎在评论区留言&#xff0c;我们一起交流学习。你的支持是我继续创作的动力&#xff01; &#x1f44d; 点赞、收藏与分享&#xff1a;觉得这篇文章对你有帮助吗&#xff1…

ElasticSearch的学习

介绍 ElasticSearch&#xff08;简称ES&#xff09;是一个开源的分布式搜索和数据分析引擎&#xff0c;是用Java开发并且是当前最流行的开源的企业级搜索引擎&#xff0c;能够达到近实时搜索&#xff0c;它专门设计用于处理大规模的文本数据和实现高性能的全文检索。 Elastic…

Y20030018基于Java+Springboot+mysql+jsp+layui的家政服务系统的设计与实现 源代码 文档

家政服务系统的设计与实现 1.摘要2.开发目的和意义3.系统功能设计4.系统界面截图5.源码获取 1.摘要 随着人们生活水平的提高&#xff0c;老龄化、少子化等多重因素影响&#xff0c;我国对家政服务人群的需求与日俱增。家政服务行业对我国的就业和社会效益贡献也与日俱增&#…

南京仁品耳鼻喉专科医院:12月启动公益义诊月

专业医疗资源送至“家门口”&#xff01;南京仁品耳鼻喉专科医院启动公益义诊月 随着2024年即将步入尾声&#xff0c;南京仁品耳鼻喉医院为回馈社会&#xff0c;提升公众健康福祉&#xff0c;将于12月隆重推出“三甲专家公益义诊月”活动。此次活动旨在通过汇聚众多耳鼻喉领域…

ospf协议(动态路由协议)

ospf基本概念 定义 OSPF 是典型的链路状态路由协议&#xff0c;是目前业内使用非常广泛的 IGP 协议之一。 目前针对 IPv4 协议使用的是 OSPF Version 2 &#xff08; RFC2328 &#xff09;&#xff1b;针对 IPv6 协议使用 OSPF Version 3 &#xff08; RFC2740 &#xff09;。…

Linux - nfs服务器

五、nfs服务器 1、基础 NFS服务器可以让PC将网络中的NFS服务器共享的目录挂载到本地端的文件系统中&#xff0c;而在本地端的系统 中看来&#xff0c;那个远程主机的目录就好像是自己的一个磁盘分区一样。 由于NFS支持的功能比较多&#xff0c;而不同的功能都会使用不同的程…

现代网络架构PCI DSS合规范围确定和网络分割措施实施探讨

本文为atsec和作者技术共享类文章&#xff0c;旨在共同探讨信息安全业界的相关话题。未经许可&#xff0c;任何单位及个人不得以任何方式或理由对本文的任何内容进行修改。转载请注明&#xff1a;atsec信息安全和作者名称 1 引言 支付卡行业数据安全标准 &#xff08;P…

第二讲:C++基础语法与程序结构

课程目标&#xff1a; 使学生掌握C的基本语法规则。教授学生如何编写简单的C程序&#xff0c;并理解程序的基本结构。通过实践&#xff0c;加深学生对C编程的理解和兴趣。 课程内容&#xff1a; 一、引言 回顾上一讲关于C的概述&#xff0c;强调学习基础语法的重要性。简要…

鸿蒙开发:自定义一个任意位置弹出的Dialog

前言 鸿蒙开发中&#xff0c;一直有个问题困扰着自己&#xff0c;想必也困扰着大多数开发者&#xff0c;那就是&#xff0c;系统提供的dialog自定义弹窗&#xff0c;无法实现在任意位置进行弹出&#xff0c;仅限于CustomDialog和Component struct的成员变量&#xff0c;这就导致…

深入浅出:开发者如何快速上手Web3生态系统

Web3作为互联网的未来发展方向&#xff0c;正在逐步改变传统互联网架构&#xff0c;推动去中心化技术的发展。对于开发者而言&#xff0c;Web3代表着一个充满机遇与挑战的新领域&#xff0c;学习和掌握Web3的基本技术和工具&#xff0c;将为未来的项目开发提供强大的支持。那么…

Q-2A型金相试样切割机

产品概述 在金相试样制备过程中&#xff0c;试样材料的切割是试样制备的首道重要工序,本机利用高速旋转的薄片砂轮来截取试样&#xff0c;适直切割较硬的金属材料&#xff0c;本机有冷却装置&#xff0c;用来带走切割时所产生的热量&#xff0c;避免试样过热而改变组织。 主要…

十二、Pod的扩缩容-手动/自动-HPA

在实际生产系统中,经常会遇到某个服务需要扩容的场景,也可能会遇到由于资源紧张或者工作负载降低而需要减少服务实例数量的场景。此时可以利用Deployment/RC的Scale机制来完成这些工作。 Kubernetes对Pod的扩缩容操作提供了手动和自动两种模式,手动模式通过运行kubectl sca…

Ubuntu环境中RocketMQ安装教程

参考教程 https://blog.csdn.net/weixin_56219549/article/details/126143231 1、安装JDK&#xff0c;并配置环境变量&#xff08;略&#xff09; 2、下载RocketMQ安装包 RocketMQ下载地址&#xff0c;选择二进制包下载 unzip rocketmq-all-5.0.0-ALPHA-bin-release.zip 使…

传输控制协议(TCP)

传输控制协议是Internet一个重要的传输层协议。TCP提供面向连接、可靠、有序、字节流传输服务。 1、TCP报文段结构 注&#xff1a;TCP默认采用累积确认机制。 2、三次握手、四次挥手 &#xff08;1&#xff09;当客户向服务器发送完最后一个数据段后&#xff0c;发送一个FIN段…

我们来学mysql -- 事务之概念(原理篇)

事务的概念 题记一个例子一致性隔离性原子性持久性 题记 在漫长的编程岁月中&#xff0c;存在一如既往地贯穿着工作&#xff0c;面试的概念这类知识点&#xff0c;事不关己当然高高挂起&#xff0c;精准踩坑时那心情也的却是日了&#x1f436;请原谅我的粗俗&#xff0c;遇到B…

2024 ccpc 辽宁省赛 E(构造 思维?)L(二分+一点点数论知识?)

E 题意&#xff1a; 可以注意到&#xff1a; 我的两种方格都四个方格的大小。 所以 如果存在一种摆放方式 那么 4|nm。 再考虑一种特殊的情况 22 &#xff0c;此时虽然我的积是4 但是无法摆放的。 1>对于 4 | n,或者 4 | m.我直接摆放第二种方格就可以了。 如果我n 是4 的…

自定义类型: 结构体、枚举 、联合

目录 结构体 结构体类型的声明 匿名结构体 结构的自引用 结构体变量的定义和初始化 结构体成员变量的访问 结构体内存对齐 结构体传参 位段 位段类型的声明 位段的内存分配 位段的跨平台问题 位段的应用 枚举 枚举类型的定义 枚举的优点 联合体(共用体) 联合…

道可云人工智能元宇宙每日资讯|第三届京西地区发展论坛成功召开

道可云元宇宙每日简报&#xff08;2024年11月27日&#xff09;讯&#xff0c;今日元宇宙新鲜事有&#xff1a; 工信部等十二部门印发《5G规模化应用“扬帆”行动升级方案》 11月25日&#xff0c;工业和信息化部等十二部门印发《5G规模化应用“扬帆”行动升级方案》。《方案》…

更多开源创新 挑战OpenAI-o1的模型出现和AI个体模拟突破

每周跟踪AI热点新闻动向和震撼发展 想要探索生成式人工智能的前沿进展吗&#xff1f;订阅我们的简报&#xff0c;深入解析最新的技术突破、实际应用案例和未来的趋势。与全球数同行一同&#xff0c;从行业内部的深度分析和实用指南中受益。不要错过这个机会&#xff0c;成为AI领…

说说Elasticsearch查询语句如何提升权重?

大家好&#xff0c;我是锋哥。今天分享关于【说说Elasticsearch查询语句如何提升权重&#xff1f;】面试题。希望对大家有帮助&#xff1b; 说说Elasticsearch查询语句如何提升权重&#xff1f; 1000道 互联网大厂Java工程师 精选面试题-Java资源分享网 在 Elasticsearch 中&…