工作中常用的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.hqwc.cn/news/255687.html

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

相关文章

Linux查看命令的绝对路径

Linux查看命令的绝对路径

linux查看命令的绝对路径 在Linux中&#xff0c;可以使用以下命令来查看命令的绝对路径&#xff1a; 1、which 命令名 例如&#xff0c;要查看chronyc命令的绝对路径&#xff0c;可以运行&#xff1a; which chronyc 2、whereis 命令名 例如&#xff0c;要查看chronyc命令…
阅读更多...
Python-关系运算符详解

Python-关系运算符详解

关系运算符&#xff1a;比较两个操作数的大小或者相等关系 < > ! 1、关系运算符的关系表达式返回值是布尔类型bool 成立就是真&#xff0c;即1&#xff1b;不成立就是假&#xff0c;即0 2、关系运算符还可以比较字符 字符根据字典序比较&#xff0c;先看首字母在…
阅读更多...
SolidWorks Simulation 有限元分析-升降架分析

SolidWorks Simulation 有限元分析-升降架分析

问题描述&#xff1a;一个载重为 1800N 的升降架承受一外部水压柱筒的作用&#xff0c;该水压柱筒与基座上的滑块相连。 1. 打开零件。 2. 我们新建一个静应力分析的算例&#xff0c;如下图所示。 3. 设置材料。我们需要为模型指定模型的材料属性。点击选中模型&#xff0c;鼠…
阅读更多...
Python超级详细的raise用法

Python超级详细的raise用法

当程序出现错误时&#xff0c;系统会自动引发异常。除此之外&#xff0c;Python 也允许程序自行引发异常&#xff0c;自行引发异常使用 raise 语句来完成。 很多时候&#xff0c;系统是否要引发异常&#xff0c;可能需要根据应用的业务需求来决定&#xff0c;如果程序中的数据…
阅读更多...
office办公技能|ppt插件使用

office办公技能|ppt插件使用

PPT插件获取&#xff1a;链接&#xff1a;https://pan.baidu.com/s/1BOmPioUKeY2TdC-1V-o3Vw 提取码&#xff1a;tdji 一、ppt插件介绍 PPT插件是一种可以帮助用户在Microsoft PowerPoint软件中添加各种额外功能和效果的应用程序。使用PPT插件可以让用户更加轻松地制作出专业、…
阅读更多...
JS中Map对象与object的区别

JS中Map对象与object的区别

若想了解Map对象可以阅读本人这篇ES6初步了解Map Map对象与object有什么区别&#xff1f;让我为大家介绍一下吧&#xff01; 共同点 二者都是以key-value的形式对数据进行存储 const obj {name:"zs",age:18}console.log(obj)let m new Map()m.set("name&quo…
阅读更多...
Linux--文件权限与shell外壳的理解

Linux--文件权限与shell外壳的理解

目录 一.Linux的用户与用户切换&#xff0c;提权 二.对文件权限的理解 1.文件权限角色的权限文件属性 2.Linux中的三种角色 3.为什么会存在所属组这个角色 4.文件属性的意义 4.1.第一个字母的意义 4.2 第2——第10个字母的意义 4.3修改文件权限的方法 三.目录权限 四…
阅读更多...
MATLAB - 绘制立体图（平面+水深）

MATLAB - 绘制立体图（平面+水深）

目录 代码结果 代码 % 在 X-Y 平面上绘图 % 正常绘制平面图 [X,Y,Z] peaks; contour(X,Y,Z,20); hold on% ****重点******************************************** % 改为三维视图&#xff0c;具体可以help % view(3); %此时的平面图对应z0 &#xff1b;默认az-37.5&#x…
阅读更多...
【精选】如何设置IDEA中包结构树状或平展，使包结构更清晰明了

【精选】如何设置IDEA中包结构树状或平展，使包结构更清晰明了

可以通过设置项目窗口的树外观对包结构&#xff0c;以及项目栏目中各类文件是否显示&#xff0c;使得自己的项目栏更加清晰&#xff0c;具体设置方法看图&#xff1a; 其中可以设置显示的成员&#xff0c;显示排除文件&#xff0c;显示可见性图标等等&#xff0c;可以通过设置平…
阅读更多...
C++STL的string类(一)

C++STL的string类(一)

文章目录 前言C语言的字符串 stringstring类的常用接口string类的常见构造string (const string& str);string (const string& str, size_t pos, size_t len npos); capacitysize和lengthreserveresizeresize可以删除数据 modify尾插插入字符插入字符串 inserterasere…
阅读更多...
python-比较Excel两列数据，并分别显示差异

python-比较Excel两列数据，并分别显示差异

利用 openpyxl 模块&#xff0c;操作Excel&#xff0c;比较Excel两列数据&#xff0c;并分别显示差异 表格数据样例如下图 A&#xff0c;B两列是需要进行比较的数据&#xff08;数据源为某网站公开数据&#xff09;&#xff1b;C&#xff0c;D两列是比较结果的输出列 A&#…
阅读更多...
elasticsearch-head 启动教程

elasticsearch-head 启动教程

D:\elasticsearch-head-master>grunt server ‘grunt’ 不是内部或外部命令&#xff0c;也不是可运行的程序 或批处理文件。 npm install -g grunt-clinpm install
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023