工作中常用的RabbitMQ实践

目录

1.前置

2.导入依赖

3.生产者

4.消费者

5.验证

验证Direct

验证Fanout

验证Topic

1.前置

安装了rabbitmq，并成功启动

2.导入依赖

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

3.生产者

生产端项目结构：
 

逻辑：生产者只对交换机进行生产，至于队列绑定等放在消费端进行执行

BusinessConfig

定义了三个不同类型的交换机

direct类型:(当生产者往该交换机发送消息时，他必须指定固定的routingkey，当routingkey值为空，他也会匹配routingkey为空的队列)

fanout类型:(当生产者往该交换机发送消息时，他所绑定的队列都会收到消息，routingkey即使写了也会忽略，一般为空字符串)

Topic类型:(当生产者往该交换机发送消息时，他并不像direct指定固定的routingkey，可以进行模糊匹配，当该routingkey为空时，他会匹配routingkey为空的队列)

package com.zsp.quartz.queue;import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;/*** @Author: ZhangSP* @Date: 2023/12/7  14:05*/
@Slf4j
@Configuration
public class BusinessConfig {// 声明direct交换机public static final String EXCHANGE_DIRECT= "exchange_direct_inform";// 声明fanout交换机public static final String EXCHANGE_FANOUT= "exchange_fanout_inform";// 声明topic交换机public static final String EXCHANGE_TOPIC= "exchange_topic_inform";
}

TestProducer

生产消息

package com.zsp.quartz.queue;import com.alibaba.fastjson.JSON;
import com.zsp.quartz.entity.User;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@SpringBootTest
@RunWith(SpringRunner.class)
public class TestProducer {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void Producer_topics_springbootTest() {//使用rabbitTemplate发送消息String message = "";User user = new User();user.setName("张三");user.setEmail("anjduahsd");message = JSON.toJSONString(user);// directrabbitTemplate.convertAndSend(BusinessConfig.EXCHANGE_DIRECT,"",message);// fanoutrabbitTemplate.convertAndSend(BusinessConfig.EXCHANGE_FANOUT,"",message);// topicrabbitTemplate.convertAndSend(BusinessConfig.EXCHANGE_TOPIC,"",message);}
}

4.消费者

消费者目录结构:

BusinessConfig:定义交换机类型，配置交换机与队列的绑定关系，通过容器工厂声明队列

package com.zsp.consumer.queue;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;import javax.annotation.PostConstruct;/*** @Author: ZhangSP* @Date: 2023/12/7  14:05*/
@Slf4j
@Configuration
public class BusinessConfig {// 声明directpublic static final String EXCHANGE_DIRECT= "exchange_direct_inform";public static final String QUEUE_DIRECT_EMAIL = "queue_direct_inform_email";public static final String QUEUE_DIRECT_SMS = "queue_direct_inform_sms";public void BindDirectEmail(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT.getType(), true);channel.queueDeclare(QUEUE_DIRECT_EMAIL, true, false, false, null);channel.queueBind(QUEUE_DIRECT_EMAIL, EXCHANGE_DIRECT, "");} catch (Exception e) {log.error("声明Direct->email队列时失败", e);}}public void BindDirectSms(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT.getType(), true);channel.queueDeclare(QUEUE_DIRECT_SMS, true, false, false, null);channel.queueBind(QUEUE_DIRECT_SMS, EXCHANGE_DIRECT, "123");} catch (Exception e) {log.error("声明Direct->sms失败", e);}}// 声明fanoutpublic static final String EXCHANGE_FANOUT= "exchange_fanout_inform";public static final String QUEUE_FANOUT_EMAIL = "queue_fanout_inform_email";public static final String QUEUE_FANOUT_SMS = "queue_fanout_inform_sms";public void BindFanoutEmail(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT.getType(), true);channel.queueDeclare(QUEUE_FANOUT_EMAIL, true, false, false, null);channel.queueBind(QUEUE_FANOUT_EMAIL, EXCHANGE_FANOUT, "");} catch (Exception e) {log.error("声明Fanout->email队列时失败", e);}}public void BindFanoutSms(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT.getType(), true);channel.queueDeclare(QUEUE_FANOUT_SMS, true, false, false, null);channel.queueBind(QUEUE_FANOUT_SMS, EXCHANGE_FANOUT,"");} catch (Exception e) {log.error("声明Fanout->sms失败", e);}}// 声明topicpublic static final String EXCHANGE_TOPIC= "exchange_topic_inform";public static final String QUEUE_TOPIC_EMAIL = "queue_topic_inform_email";public static final String QUEUE_TOPIC_SMS = "queue_topic_inform_sms";public static final String ROUTINGKEY_EMAIL="inform.#.email.#";public static final String ROUTINGKEY_SMS="inform.#.sms.#";public void BindTopicEmail(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC.getType(),true);channel.queueDeclare(QUEUE_TOPIC_EMAIL, true, false, false, null);channel.queueBind(QUEUE_TOPIC_EMAIL, EXCHANGE_TOPIC, ROUTINGKEY_EMAIL);} catch (Exception e) {log.error("声明Topic->email队列时失败", e);}}public void BindTopicSms(Channel channel) {try {channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC.getType(),true);channel.queueDeclare(QUEUE_TOPIC_SMS, true, false, false, null);channel.queueBind(QUEUE_TOPIC_SMS, EXCHANGE_TOPIC,"");} catch (Exception e) {log.error("声明Topic->sms失败", e);}}// 声明队列@Autowired@Qualifier(value = "zspConnectionFactory")private ConnectionFactory connectionFactory;@PostConstructpublic void shengmingQueue() {try {Connection connection = connectionFactory.createConnection();Channel channel = connection.createChannel(false);BindDirectEmail(channel);BindDirectSms(channel);BindFanoutEmail(channel);BindFanoutSms(channel);BindTopicEmail(channel);BindTopicSms(channel);} catch (Exception e) {log.error("业务实例声明绑定队列报错:",e);}}
}

RabbitFactory:自定义容器工厂

package com.zsp.consumer.queue;import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
@EnableRabbit
public class RabbitFactory {@Bean("zspConnectionFactory")public ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();// 设置RabbitMQ的连接信息，如主机名、端口号、用户名和密码等connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("root");connectionFactory.setPassword("root");return connectionFactory;}@Bean("rabbitListenerContainerFactory")public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(@Qualifier("zspConnectionFactory") ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setConcurrentConsumers(5);factory.setMaxConcurrentConsumers(10);return factory;}
}

ReceiveHandler:队列监听

package com.zsp.consumer.queue;import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class ReceiveHandler {//监听自定义的Direct队列@RabbitListener(queues = BusinessConfig.QUEUE_DIRECT_SMS, containerFactory = "rabbitListenerContainerFactory")public void directSMS(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Direct队列->sms队列" + jsonObject);}@RabbitListener(queues = BusinessConfig.QUEUE_DIRECT_EMAIL, containerFactory = "rabbitListenerContainerFactory")public void directEmail(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Direct队列->email队列" + jsonObject);}//监听自定义的Fanout队列@RabbitListener(queues = BusinessConfig.QUEUE_FANOUT_SMS, containerFactory = "rabbitListenerContainerFactory")public void FanoutSMS(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Fanout队列->sms队列" + jsonObject);}@RabbitListener(queues = BusinessConfig.QUEUE_FANOUT_EMAIL, containerFactory = "rabbitListenerContainerFactory")public void FanoutEmail(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Fanout队列->email队列" + jsonObject);}//监听自定义的Topic队列@RabbitListener(queues = BusinessConfig.QUEUE_TOPIC_SMS, containerFactory = "rabbitListenerContainerFactory")public void TopicSMS(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Topic队列->sms队列" + jsonObject);}@RabbitListener(queues = BusinessConfig.QUEUE_TOPIC_EMAIL, containerFactory = "rabbitListenerContainerFactory")public void TopicEmail(String msg, Message message, Channel channel) {JSONObject jsonObject = JSONObject.parseObject(msg);System.out.println("Topic队列->email队列" + jsonObject);}
}

5.验证

先启动消费者端，然后执行TestProducer

验证Direct

1.向routingkey为空的队列发消息

我们在消费者端配置了routingkey为空的队列，叫做 QUEUE_DIRECT_EMAIL

因此会打印出下面这条记录

2.向routingkey为123的队列发消息

我们在消费者端配置了routingkey为123的队列，叫做 QUEUE_DIRECT_SMS

因此会打出下面这条记录

验证Fanout

谁跟我绑定了，我都发

验证Topic

模糊匹配routingkey

匹配sms队列

会把下面这个打印出来

需要注意的是如果我们没有自定义容器工厂的话，这个containerFactory可以不写

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

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

相关文章

uni-app 微信小程序之好看的ui登录页面（五）

uni-app 微信小程序之好看的ui登录页面（五）

文章目录 1. 页面效果2. 页面样式代码 更多登录ui页面 uni-app 微信小程序之好看的ui登录页面&#xff08;一&#xff09; uni-app 微信小程序之好看的ui登录页面&#xff08;二&#xff09; uni-app 微信小程序之好看的ui登录页面&#xff08;三&#xff09; uni-app 微信小程…
阅读更多...
Ray构建GPU隔离的机器学习平台

Ray构建GPU隔离的机器学习平台

Ray框架介绍 Ray 是一个开源分布式计算框架，在 机器学习基础设施中发挥着至关重要的作用。Ray 促进分布式机器学习训练，使机器学习从业者能够有效利用多个 GPU 的能力。 Ray可以在集群上分布式地运行任务，并且可以指定任务运行时需要使用的GPU数量。Ray可与Nvidia-docker等…
阅读更多...
异常检测 | MATLAB实现基于支持向量机和孤立森林的数据异常检测(结合t-SNE降维和DBSCAN聚类)

异常检测 | MATLAB实现基于支持向量机和孤立森林的数据异常检测(结合t-SNE降维和DBSCAN聚类)

异常检测 | MATLAB实现基于支持向量机和孤立森林的数据异常检测(结合t-SNE降维和DBSCAN聚类) 目录 异常检测 | MATLAB实现基于支持向量机和孤立森林的数据异常检测(结合t-SNE降维和DBSCAN聚类)效果一览基本介绍模型准备模型设计参考资料 效果一览 基本介绍 提取有用的特征&…
阅读更多...
Java的三种代理模式实现

Java的三种代理模式实现

代理模式的定义&#xff1a; Provide a surrogate or placeholder for another object to control access to it.&#xff08;为其他对象提供一种代理以控制对这个对象的访问。&#xff09; 简单说&#xff0c;就是设置一个中间代理来控制访问原目标对象&#xff0c;达到增强原…
阅读更多...
领域驱动架构（DDD）建模

领域驱动架构（DDD）建模

一、背景 常见的软件开发方式是拿到产品需求后&#xff0c;直接考虑数据库中表应该如何设计&#xff0c;这种方式已经将设计与业务需求脱节&#xff0c;而更多的是直接考虑应该如何实现了&#xff0c;这有点本末倒置。而DDD是从领域(问题域)为出发点进行的设计方法。 领域驱动…
阅读更多...
记录 | centos源码编译bazel

记录 | centos源码编译bazel

tensorflow的源码编译依赖于 bazel 这里进行 bazel 的源码编译 1、安装依赖 sudo yum install -y java-11-openjdk sudo yum install -y java-11-openjdk-devel sudo yum install -y protobuf-compiler zip unzip2、知悉要安装的 bazel 的版本 务必安装受支持的 Bazel 版本…
阅读更多...
Linux下c开发

Linux下c开发

编程环境 Linux 下的 C 语言程序设计与在其他环境中的 C 程序设计一样&#xff0c; 主要涉及到编辑器、编译链接器、调试器及项目管理工具。编译流程 编辑器 Linux 中最常用的编辑器有 Vi。编译连接器 编译是指源代码转化生成可执行代码的过程。在 Linux 中&#xff0c;最常用…
阅读更多...
vs2022 winform 使用LiveCharts.Wpf控件出现黑框 去除方法

vs2022 winform 使用LiveCharts.Wpf控件出现黑框 去除方法

调试—>选项—>XAML热重载 去除
阅读更多...
网络安全行业大模型调研总结

网络安全行业大模型调研总结

随着人工智能技术的发展&#xff0c;安全行业大模型SecLLM&#xff08;security Large Language Model&#xff09;应运而生&#xff0c;可应用于代码漏洞挖掘、安全智能问答、多源情报整合、勒索情报挖掘、安全评估、安全事件研判等场景。 参考&#xff1a; 1、安全行业大模…
阅读更多...
uniapp 打开文件管理器上传(H5、微信小程序、android app三端)文件

uniapp 打开文件管理器上传(H5、微信小程序、android app三端)文件

H5跟安卓APP 手机打开的效果图&#xff1a; Vue页面&#xff1a; <template><view class"content"><button click"uploadFiles">点击上传</button></view> </template><script>export default {data() {return…
阅读更多...
web：[GXYCTF2019]BabyUpload（文件上传、一句话木马、文件过滤）

web：[GXYCTF2019]BabyUpload（文件上传、一句话木马、文件过滤）

题目 页面显示为文件上传 随便上传一个文件看看 上传一个文本文件显示 上传了一个图片显示 上传包含一句话木马的图片 上传了一个包含php一句话木马的文件&#xff0c;显示如上 换一个写法 上传成功 尝试上传.htaccess&#xff0c;上传失败&#xff0c;用抓包修改文件后缀 …
阅读更多...
LangChain 23 Agents中的Tools用于增强和扩展智能代理agent的功能

LangChain 23 Agents中的Tools用于增强和扩展智能代理agent的功能

LangChain系列文章 LangChain 实现给动物取名字&#xff0c;LangChain 2模块化prompt template并用streamlit生成网站 实现给动物取名字LangChain 3使用Agent访问Wikipedia和llm-math计算狗的平均年龄LangChain 4用向量数据库Faiss存储&#xff0c;读取YouTube的视频文本搜索I…
阅读更多...
Apache Flink（四）：Flink 其他实时计算框架对比

Apache Flink（四）：Flink 其他实时计算框架对比

🏡 个人主页：IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客 🚩 私聊博主：加入大数据技术讨论群聊，获取更多大数据资料。 🔔 博主个人B栈地址：豹哥教你大数据的个人空间-豹哥教你大数据个人主页-哔哩哔哩视频 根据前文描述我们知道Flink主要处…
阅读更多...
LNMP网站架构分布式搭建部署（编译安装）

LNMP网站架构分布式搭建部署（编译安装）

目录 一、数据库编译安装 二、nginx编译安装 三、php编译安装 三、通过nfs将三台不同的主机资源共享 四、基础测试 五、完成WordPress站点部署 六、完成bbs论坛站点部署 一、数据库编译安装 1、先下载安装包到/opt目录中&#xff0c;最好选择mysql-boost-5.7.44.tar.gz版…
阅读更多...
步进电机驱动芯片TB6600HG部分翻译

步进电机驱动芯片TB6600HG部分翻译

有些参数没看懂。一边设计&#xff0c;一边修正。 目录 1.芯片梗概 2.引脚配置 1)引脚含义 2)内部逻辑 3.功能详解 1&#xff09;励磁模式设置 2&#xff09;功能设置 3&#xff09;初始模式 4&#xff09;100% 电流设置(电流值) 5&#xff09;OSC 6&#xff09;衰减…
阅读更多...
python 使用 AppiumService 类启动appium server

python 使用 AppiumService 类启动appium server

一、前置说明 在Appium的1.6.0版本中引入了AppiumService类&#xff0c;可以很方便的通过该类来管理Appium服务器的启动和停止。 二、操作步骤 import osfrom appium.webdriver.appium_service import AppiumService as OriginalServerfrom libs import pathclass AppiumSer…
阅读更多...
windows判断exe应用程序是否在使用的bat脚本

windows判断exe应用程序是否在使用的bat脚本

脚本 REM 查询进程是否存在 tasklist|findstr /i "mysqld.exe">nul &&echo y >2.log ||echo n >2.log REM 读取文本内容赋值给变量 set /P resu<2.log if %resu% y (echo process in use ) else (echo process not in use )我们已mysqld.exe…
阅读更多...
C++ day58 每日温度 下一个更大元素

C++ day58 每日温度 下一个更大元素

题目1&#xff1a;739 每日温度 题目链接&#xff1a;每日温度 对题目的理解 temperature[i]表示每天的温度&#xff0c;返回数组answer&#xff0c;answer[i]指对于第i天&#xff0c;下一个更高温度最近出现在几天后&#xff0c;如果气温在这之后都不会升高&#xff0c;用0…
阅读更多...
720度vr虚拟家居展厅提升客户的参观兴致

720度vr虚拟家居展厅提升客户的参观兴致

VR虚拟展厅线上3D交互展示的优势有以下几点&#xff1a; 打破了场馆的展示限制&#xff0c;可展示危险性制品、珍贵稀有物品、超大型设备等&#xff0c;同时提供了更大的展示空间和更丰富的展示内容。 可提供企业真实环境的实时VR全景参观&#xff0c;提升潜在客户信任度。 提供…
阅读更多...
JAVA导出PDF（一）

JAVA导出PDF（一）

思路一&#xff1a;直接导出pdf 使用freemarker和itext把html转pdf&#xff0c;存在中文乱码&#xff0c;宋体不识别&#xff0c;需下载simsun.ttc字体&#xff0c;空白占位符不识别等等问题&#xff0c;对前端依赖性较大&#xff0c;不推荐&#xff1b; 适用范围&#xff1a;…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023