腻害,高人都是这样玩SpringBoot整合RabbitMQ
一、认识 RabbitMQRabbitMQ 简介以 AMQP 协议:(1)RabbitMQ 是开源的消息代理和队列服务器,用来通过普通协议在完全不同的应用之间共享数据,RabbitMQ 底层是用了 Erlang 语言来编写的,并且 RabbitMQ 是基于 AMQP 协议的.(2)RabbitMQ 不仅仅可以使用 java 客户端进行编写,且可以使用其他的语言(python,php等…),它提供了
一、认识 RabbitMQ
RabbitMQ 简介以 AMQP 协议:
(1)RabbitMQ 是开源的消息代理和队列服务器,用来通过普通协议在完全不同的应用之间共享数据,RabbitMQ 底层是用了 Erlang 语言来编写的,并且 RabbitMQ 是基于 AMQP 协议的.
(2)RabbitMQ 不仅仅可以使用 java 客户端进行编写,且可以使用其他的语言(python,php等…),它提供了丰富的API
RabbitMQ 的优点:
(1)开源,性能优秀,稳定性保障
(2)与 SpringAMQP 完美的整合,API 丰富 (Spring基于 RabbitMQ 提供了一套框架,叫做AMQP 框架)这套框架不仅提供了原生的 RabbitMQ,而且还提供了丰富可扩张的API帮助开发人员更好的去应用
(3)集群模式丰富,表达式配置,HA模式,镜像队列模型
说明:(保证数据不丢失的提前做到高可靠性,可用性)普遍使用的镜像队列模式
(4)AMQP 全称:Advanced Message Queuing Protocl AMQP 翻译过来:高级消息队列协议
二、安装 RabbitMQ
这里为了方便在 docker 上安装 RabbitMQ
(1)首先搜索 RabbitMQ 的安装命令:https://hub.docker.com/_/rabbitmq
docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management
1
安装成功:
对于映射的端口需要注意两个:15672 和 5672,其中 15672 是管理的端口,5672 是通讯的端口
可以看到docker容器 15672 对应的主机映射端口是 32771,
默认的用户名和密码都是 guest
三、SpringBoot 整合 RabbitMQ
1. 加入依赖
2. 配置 application.properties
注意这里的端口选择 32781,对应rabbitmq 容器的端口是 5672,选择通讯端口,切忌不用选择管理端口
spring.rabbitmq.host=192.168.245.133
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=32781
1234
3. 直接交换模式
直接交换模式可以参考:https://blog.csdn.net/fakerswe/article/details/81508963
所谓“直接连接交换机”就是:Producer(生产者)投递的消息被DirectExchange (交换机)转发到通过routingkey绑定到具体的某个Queue(队列),把消息放入队列,然后Consumer从Queue中订阅消息
这里的具体某个队列是由参数 routingKey 控制的,消息是通过这个参数往队列上发的
RabbitMQ消息模型的核心思想(core idea): 生产者会把消息发送给RabbitMQ的交换中心(Exchange),Exchange的一侧是生产者,另一侧则是一个或多个队列,由Exchange决定一条消息的生命周期–发送给某些队列,或者直接丢弃掉。
(1)配置 RabbitDirectConfig
@Configuration
public class RabbitDirectConfig {
public final static String DIRECTNAME = "yolo-direct";
// 消息队列
@Bean
Queue queue() {
return new Queue("hello.yolo");
}
@Bean
DirectExchange directExchange() {
return new DirectExchange(DIRECTNAME, true, false);
}
// 将 queue 和 directExchange 绑定到一起
@Bean
Binding binding() {
return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
}
}
123456789101112131415161718192021
(2)消费者
@Component
public class DirectReceiver {
// 监听队列
@RabbitListener(queues = "hello.yolo")
public void handler1(String msg) {
System.out.println("handler1>>>" + msg);
}
}
12345678
(3)测试:生产者
@Autowired
RabbitTemplate rabbitTemplate;
@Test
public void contextLoads() {
//将消息转发到 routingKey 为 hello.yolo 的队列,对应 DirectReceiver 的监听队列名
rabbitTemplate.convertAndSend("hello.yolo", "hello yolo! ni hao!");
}
1234567
4. 广播模式
可以参考:https://blog.csdn.net/fakerswe/article/details/81455340
简单的讲,就是把交换机(Exchange)里的消息发送给所有绑定该交换机的队列,忽略routingKey。
由图可知,生产者把消息发送到交换机后,由交换机发送给消费者队列。消费者队列如果想要接收到交换机里的消息,那么需要保证:队列绑定的交换机名称要和交换机一致,这个是广播模式的关键,也是MQ后续所有模式最粗略的前提。
这里消息是通过生产者发往交换机的,然后交换机再发送给绑定的队列
(1)配置广播模式
@Configuration
public class RabbitFanoutConfig {
public static final String FANOUTNAME = "yolo-fanout";
/**
* 队列1
* @return
*/
@Bean
Queue queueOne() {
return new Queue("queue-one");
}
/**
* 队列2
* @return
*/
@Bean
Queue queueTwo() {
return new Queue("queue-two");
}
/**
* 交换机
* @return
*/
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange(FANOUTNAME, true, false);
}
/**
* 绑定队列1
* @return
*/
@Bean
org.springframework.amqp.core.Binding bindingOne() {
return BindingBuilder.bind(queueOne()).to(fanoutExchange());
}
/**
* 绑定队列2
* @return
*/
@Bean
Binding bindingTwo() {
return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
}
}
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
(2)消费者
/**
* 定义接收器:消费者
*/
@Component
public class FanoutReceiver {
/**
* 接收消息队列1
* @param msg
*/
@RabbitListener(queues = "queue-one")
public void handler1(String msg) {
System.out.println("FanoutReceiver:handler1:" + msg);
}
/**
* 接收消息队列2
* @param msg
*/
@RabbitListener(queues = "queue-two")
public void handler2(String msg) {
System.out.println("FanoutReceiver:handler2:" + msg);
}
}
12345678910111213141516171819202122
(3)测试:生产者
/**
* 往交换机上发送信息
*/
@Test
public void test1() {
rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null,"hello fanout!");
}
1234567
这里需要注意,需要先启动消费者,再启动生产者,否则先启动生产者,exchange接收到消息后发现没有队列对其感兴趣,就会将消息丢掉,此时跟 routingKey 无关
队列1和队列2 均收到了消息
5. 主题路由匹配模式
可参考:https://blog.csdn.net/weixin_43770545/article/details/90902788
假如你想在淘宝上买一双运动鞋,那么你是不是会在搜索框中搜“XXX运动鞋”,这个时候系统将会模糊匹配的所有符合要求的运动鞋,然后展示给你。
所谓“主题路由匹配交换机”也是这样一个道理,但是使用时也有一定的规则。
String routingkey = “testTopic.#”;
String routingkey = “testTopic.*”;
12
* 表示只匹配一个词
#表示匹配多个词
(1)配置 topic 模式
@Configuration
public class RabbitTopicConfig {
public static final String TOPICNAME = "yolo-topic";
@Bean
TopicExchange topicExchange() {
return new TopicExchange(TOPICNAME, true, false);
}
@Bean
Queue xiaomi() {
return new Queue("xiaomi");
}
@Bean
Queue huawei() {
return new Queue("huawei");
}
@Bean
Queue phone() {
return new Queue("phone");
}
@Bean
Binding xiaomiBinding() {
//xiaomi.# 表示如果路由的 routingKey 是以xiaomi 开头就会发送到 xiaomi 这个队列上
return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
}
@Bean
Binding huaweiBinding() {
//huawei.#
return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
}
@Bean
Binding phoneBinding() {
// #.phone.# 表示routingKey 中包含 phone 就会被发送到 phone 这个队列上
return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
}
}
123456789101112131415161718192021222324252627282930313233343536373839404142
(2) 消费者
@Component
public class TopicReceiver {
@RabbitListener(queues = "xiaomi")
public void handler1(String msg) {
System.out.println("TopicReceiver:handler1:" + msg);
}
@RabbitListener(queues = "huawei")
public void handler2(String msg) {
System.out.println("TopicReceiver:handler2:" + msg);
}
@RabbitListener(queues = "phone")
public void handler3(String msg) {
System.out.println("TopicReceiver:handler3:" + msg);
}
}
1234567891011121314151617
(3)测试:生产者
@Test
public void test2() {
//可以被小米的队列收到
rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "xiaomi.news", "小米新闻");
//可以被手机的队列收到
rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "vivo.phone", "vivo 手机");
//可以被华为和手机的队列收到
rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "huawei.phone", "华为手机");
}
123456789
6. header 模式
这种模式使用的是 header 中的 key/value (键值对) 匹配队列,也和 routingKey 无关
(1)配置 config
@Configuration
public class RabbitHeaderConfig {
public static final String HEADERNAME = "yolo-header";
@Bean
HeadersExchange headersExchange() {
return new HeadersExchange(HEADERNAME, true, false);
}
@Bean
Queue queueName() {
return new Queue("name-queue");
}
@Bean
Queue queueAge() {
return new Queue("age-queue");
}
@Bean
Binding bindingName() {
Map<String, Object> map = new HashMap<>();
//
map.put("name", "yolo");
//whereAny 表示消息的header中只要有一个header匹配上map中的key,value,就把消息发送到对应的队列上
return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();
}
@Bean
Binding bindingAge() {
//只要有,age 这个字段,就发送到相应的队列上去
return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();
}
}
1234567891011121314151617181920212223242526272829303132333435
(2)消费者
@Component
public class HeaderReceiver {
@RabbitListener(queues = "name-queue")
public void handler1(byte[] msg) {
System.out.println("HeaderReceiver:handler1:" + new String(msg, 0, msg.length));
}
@RabbitListener(queues = "age-queue")
public void handler2(byte[] msg) {
System.out.println("HeaderReceiver:handler2:" + new String(msg, 0, msg.length));
}
}
123456789101112
(3)测试
@Test
public void test3() {
//对应 RabbitHeaderConfig 中的map 的 key / value
Message nameMsg = MessageBuilder.withBody("hello yolo !".getBytes()).setHeader("name","yolo").build();
Message ageMsg = MessageBuilder.withBody("hello 99 !".getBytes()).setHeader("age","99").build();
//此时发送的消息接收,跟 routingKey无关,跟消息的 header 内容有关
rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);
rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);
}
123456789
如果更改,header 里的键值对:则无法匹配成功,队列收不到该信息
更多推荐
所有评论(0)