rabbitmq 学习三-spring-boot中配置交换器和队列

2024-03-14 21:48
文章标签 java 配置 学习 spring 队列 boot rabbitmq 交换器

本文主要是介绍rabbitmq 学习三-spring-boot中配置交换器和队列，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

rabbitmq主要的交换器类型有fanout、direct、topic、headers
下面分别介绍三种常用的交换器使用方法
1.fanout交换器
将所有发送到该交换器的消息会路由到所有与该交换器绑定的队列中
2.direct交换器
将消息路由到RoutingKey完全匹配的队列中
3.topic 交换器
将消息路由到RoutingKey匹配的队列中，匹配的规则支持特殊字符 ”*“和“#”
一、fanout交换器使用和配置
1.声明队列、交换器并将队列绑定到交换器

@Configuration
public class RabbitFanoutExchangeConfiguration {/*** 声明队列* @return*/@Bean(name = "q.test1")public Queue queue() {return new Queue("q.test1");}@Bean("q.test2")public Queue queue2() {return new Queue("q.test2");}/*** 声明交换器* @return*/@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("x.test1");}/*** 绑定队列到交换器* @return*/@Beanpublic Binding bindingQueue2Exchange() {return BindingBuilder.bind(queue()).to(fanoutExchange());}@Beanpublic Binding bindingQueue2FanoutExchange() {return BindingBuilder.bind(queue2()).to(fanoutExchange());}
}

2.创建消息生产者

@Component
public class RabbitProducer {@AutowiredAmqpTemplate amqpTemplate;public void sendMessage(Object object) {try {Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8")).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID().toString()).build();// 指定exchange 为x.test1amqpTemplate.send("x.test1","",message);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}

3.创建消息消费者

@Component
@RabbitListener(queues = "q.test1")
public class FanoutExchangeConsumer1 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.test1------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}@Component
@RabbitListener(queues = "q.test2")
public class FanoutExchangeConsumer2 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.test2------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}

二、direct交换器使用和配置
1.声明队列、交换器并将队列绑定到交换器

@Configuration
public class RabbitDirectExchangeConfiguration {/*** 动态声明队列* @return*/@Bean(name = "q.direct1")public Queue queue1() {return new Queue("q.direct1");}/*** 动态声明队列* @return*/@Bean(name = "q.direct2")public Queue queue2() {return new Queue("q.direct2");}/*** 动态声明交换器* @return*/@Beanpublic DirectExchange directExchange() {return new DirectExchange("x.direct");}/*** 使用路由键r.direct.routingKey1将交换器绑定到队列 q.direct1* @return*/@Beanpublic Binding bindingQueue1Exchange() {return BindingBuilder.bind(queue1()).to(directExchange()).with("r.direct.routingKey1");}/*** 使用路由键r.direct.routingKey2 将交换器绑定到队列 q.direct1* @return*/@Beanpublic Binding bindingExchange2Queue2() {return BindingBuilder.bind(queue1()).to(directExchange()).with("r.direct.routingKey2");}/*** 使用路由键 r.direct.routingKey1将交换器绑定到队列 q.direct2* @return*/@Beanpublic Binding bindingExchange2Queue() {return BindingBuilder.bind(queue2()).to(directExchange()).with("r.direct.routingKey1");}
}

2.创建消息生产者

public void sendMessage2DirectExchange(Object object) {try {Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8")).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID().toString()).build();// 指定exchange 为x.direct, 路由键为 r.direct.routingKey1amqpTemplate.send("x.direct","r.direct.routingKey1",message);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}public void sendMessage2DirectDirectExchange(Object object) {try {Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8")).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID().toString()).build();// 指定exchange 为x.direct, 路由键为 r.direct.routingKey2amqpTemplate.send("x.direct","r.direct.routingKey2",message);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}

3.创建消息消费者

@Component
@RabbitListener(queues = "q.direct1")
public class DirectExchangeConsumer1 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.direct1------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}@Component
@RabbitListener(queues = "q.direct2")
public class DirectExchangeConsumer2 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.direct2------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}

三、topic交换器配置和使用
1.声明队列、交换器并将队列绑定到交换器

@Configuration
public class TopicExchangeConfiguration {/*** 动态声明队列* @return*/@Beanpublic Queue topicQueue1() {return new Queue("q.topic1");}/*** 动态声明队列* @return*/@Beanpublic Queue topicQueue2() {return new Queue("q.topic2");}/*** 动态声明交换器* @return*/@Beanpublic TopicExchange topicExchange() {return new TopicExchange("x.topic");}@Beanpublic Binding bindingTopicExchange2Queue1() {return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("*.*.test");}@Beanpublic Binding bindingTopicExchange2Queue2() {return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("*.topic.*");}@Beanpublic Binding bindingTopicExchange2Queue3() {return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("com.#");}
}

2.创建消息生产者

public void sendMessage2TopicMessage(Object object) {try {Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8")).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID().toString()).build();// 指定exchange 为x.direct, 路由键为 r.direct.routingKey2amqpTemplate.send("x.topic","com.test",message);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}public void sendMessage2TopicMessage2(Object object) {try {Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8")).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("UTF-8").setMessageId(UUID.randomUUID().toString()).build();// 指定exchange 为x.direct, 路由键为 r.direct.routingKey2amqpTemplate.send("x.topic","client.topic.test",message);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}

3.创建消息消费者

@Component
@RabbitListener(queues = "q.topic1")
public class TopicExchangeConsumer1 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.topic1------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}@Component
@RabbitListener(queues = "q.topic2")
public class TopicExchangeConsumer2 {@RabbitHandlerpublic void receiveMessage(Object object) {Message message=(Message)object;byte bytes[]=null;if (message != null) {bytes=message.getBody();}try {String msg=new String(bytes,"UTF-8");System.out.println("q.topic2------------------------"+msg);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}
}

四、创建消息发送测试controller

@RestController
@RequestMapping(value = "/messages")
public class MessageController {@AutowiredRabbitProducer rabbitProducer;@PostMapping(value = "/fanout")public Map<String,Object> sendMessage() {Map<String,Object> map=new HashMap<>();map.put("username","test1");map.put("password","123456");map.put("name","我是fanout 交换器测试人员");rabbitProducer.sendMessage(map);Map<String,Object> resultMap=new HashMap<>();resultMap.put("code","200");resultMap.put("message","success");return resultMap;}@PostMapping(value = "/direct")public Map<String,Object> sendMessage2DirectExchange() {Map<String,Object> map=new HashMap<>();map.put("username","test2");map.put("password","123456");map.put("name","我是direct 交换器测试人员");rabbitProducer.sendMessage2DirectExchange(map);Map<String,Object> resultMap=new HashMap<>();resultMap.put("code","200");resultMap.put("message","success");return resultMap;}@PostMapping(value = "/topic")public Map<String,Object> sendMessage2TopicExchange() {Map<String,Object> map=new HashMap<>();map.put("username","test3");map.put("password","123456");map.put("name","我是topic 交换器测试人员");rabbitProducer.sendMessage2TopicMessage(map);Map<String,Object> resultMap=new HashMap<>();resultMap.put("code","200");resultMap.put("message","success");return resultMap;}@PostMapping(value = "/topic1")public Map<String,Object> sendMessage2TopicExchange1() {Map<String,Object> map=new HashMap<>();map.put("username","test3");map.put("password","123456");map.put("name","我是topic 交换器测试人员");rabbitProducer.sendMessage2TopicMessage2(map);Map<String,Object> resultMap=new HashMap<>();resultMap.put("code","200");resultMap.put("message","success");return resultMap;}
}

源码下载地址
https://github.com/tangyajun/spring-boot-rabbit-consumer
https://github.com/tangyajun/rabbitmq-spring-demo

这篇关于rabbitmq 学习三-spring-boot中配置交换器和队列的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/809812。

相关文章

Spring Boot @RestControllerAdvice全局异常处理最佳实践

Spring Boot @RestControllerAdvice全局异常处理最佳实践

《SpringBoot@RestControllerAdvice全局异常处理最佳实践》本文详解SpringBoot中通过@RestControllerAdvice实现全局异常处理,强调代码复用、统... 目录前言一、为什么要使用全局异常处理？二、核心注解解析1. @RestControllerAdvice2
阅读更多...
Spring IoC 容器的使用详解(最新整理)

Spring IoC 容器的使用详解(最新整理)

《SpringIoC容器的使用详解(最新整理)》文章介绍了Spring框架中的应用分层思想与IoC容器原理,通过分层解耦业务逻辑、数据访问等模块,IoC容器利用@Component注解管理Bean... 目录1. 应用分层2. IoC 的介绍3. IoC 容器的使用3.1. bean 的存储3.2. 方法注
阅读更多...
Spring事务传播机制最佳实践

Spring事务传播机制最佳实践

《Spring事务传播机制最佳实践》Spring的事务传播机制为我们提供了优雅的解决方案,本文将带您深入理解这一机制,掌握不同场景下的最佳实践,感兴趣的朋友一起看看吧... 目录1. 什么是事务传播行为2. Spring支持的七种事务传播行为2.1 REQUIRED（默认）2.2 SUPPORTS2
阅读更多...
怎样通过分析GC日志来定位Java进程的内存问题

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》：本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.
阅读更多...
Java进程异常故障定位及排查过程

Java进程异常故障定位及排查过程

《Java进程异常故障定位及排查过程》：本文主要介绍Java进程异常故障定位及排查过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、故障发现与初步判断1. 监控系统告警2. 日志初步分析二、核心排查工具与步骤1. 进程状态检查2. CPU 飙升问题3. 内存
阅读更多...
java中新生代和老生代的关系说明

java中新生代和老生代的关系说明

《java中新生代和老生代的关系说明》：本文主要介绍java中新生代和老生代的关系说明,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、内存区域划分新生代老年代二、对象生命周期与晋升流程三、新生代与老年代的协作机制1. 跨代引用处理2. 动态年龄判定3. 空间分
阅读更多...
Java设计模式---迭代器模式(Iterator)解读

Java设计模式---迭代器模式(Iterator)解读

《Java设计模式---迭代器模式(Iterator)解读》：本文主要介绍Java设计模式---迭代器模式(Iterator),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录1、迭代器（Iterator）1.1、结构1.2、常用方法1.3、本质1、解耦集合与遍历逻辑2、统一
阅读更多...
Java内存分配与JVM参数详解(推荐)

Java内存分配与JVM参数详解(推荐)

《Java内存分配与JVM参数详解(推荐)》本文详解JVM内存结构与参数调整,涵盖堆分代、元空间、GC选择及优化策略,帮助开发者提升性能、避免内存泄漏,本文给大家介绍Java内存分配与JVM参数详解,... 目录引言JVM内存结构JVM参数概述堆内存分配年轻代与老年代调整堆内存大小调整年轻代与老年代比例元空
阅读更多...
深度解析Java DTO(最新推荐)

深度解析Java DTO(最新推荐)

《深度解析JavaDTO(最新推荐)》DTO（DataTransferObject）是一种用于在不同层（如Controller层、Service层）之间传输数据的对象设计模式,其核心目的是封装数据,... 目录一、什么是DTO？DTO的核心特点：二、为什么需要DTO？（对比Entity）三、实际应用场景解析
阅读更多...
Java 线程安全与 volatile与单例模式问题及解决方案

Java 线程安全与 volatile与单例模式问题及解决方案

《Java线程安全与volatile与单例模式问题及解决方案》文章主要讲解线程安全问题的五个成因（调度随机、变量修改、非原子操作、内存可见性、指令重排序）及解决方案,强调使用volatile关键字... 目录什么是线程安全线程安全问题的产生与解决方案线程的调度是随机的多个线程对同一个变量进行修改线程的修改操
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2