javarabbitmq错误监听_SpringBoot集成RabbitMQ实战

一、Rabbit MQ简介

Rabbit MQ是基于Erlang语言开发的消息队列中间件系统&＃xff0c;主要用来解决数据实时同步和响应问题。消息中间件产品比较常见的有ActiveMQ、Kafka、Rabbit MQ、RocketMQ等等。Rabbit MQ的优势是通过交换机绑定队列&＃xff0c;消息支持持久化&＃xff0c;服务器宕机消息仍然存在。并且Rabbit MQ支持ACK确认机制&＃xff0c;消费确认&＃xff0c;也支持分布式事务。

二、应用场景

Rabbit MQ支持多种语言客户端&＃xff0c;比如C、Ruby、Java、Python、C#、PHP等主流语言。笔者曾经做过消息中转推送中心&＃xff0c;基于Rabbit MQ的一个二次开发&＃xff0c;将微信推送消息、短信平台、语音消息、邮件等消息推送封装在消息中心&＃xff0c;通过Rabbit MQ开放接口&＃xff0c;其他系统只需要按照接口协议封装实体类发送JSON数据&＃xff0c;消息中心就能推送微信、短信、语音、邮件给用户&＃xff0c;并且具有重试机制&＃xff0c;不成功的消息记录到消息日志表中。

三、主要组件

虚拟主机&＃xff1a;RabbitMQ支持权限控制&＃xff0c;但是最小控制粒度为虚拟主机。一个虚拟主机可以包含多个交换机、队列、绑定。

交换机&＃xff1a;交换机用来绑定队列&＃xff0c;用于消息中间权限控制。

队列&＃xff1a;队列用来存放消息&＃xff0c;原理基于数据结构里面的队列。

然后通过将交换机和队列绑定在一起从而组成消息广播、消息存储与发送。

四、Spring Boot与Rabbit MQ整合搭建

新建一个Maven工程&＃xff0c;与Spring Boot整合的Rabbit MQ在pom.xml中的配置为:

org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-amqp org.springframework.boot spring-boot-starter-test test

在src/main/resources目录下新建application.yml指定项目端口号与连接Rabbit MQ配置&＃xff0c;如下:

#访问端口server: port: 8482spring: application: name: ocai-core-rabbitmq session: store-type: none rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest

新建com.ocai.core.mq.model包&＃xff0c;在该包下新建实体User&＃xff0c;代码为:

新建com.hxdi.core.mq.rabbit包&＃xff0c;包下新建RabbitConfig.java进行绑定队列&＃xff0c;注意注解&＃64;Configuration

新建com.ocai.core.mq.rabbit.hello包&＃xff0c;包下新建消息发送Bean组件&＃xff0c;如图

建立队列监听者&＃xff0c;&＃64;RabbitListener(queues &＃61; "hello")表示监听hello这个队列

新建Junit测试工具类&＃xff0c;通过调用HelloSender发送消息&＃xff0c;如图

右键HelloTest运行测试工具&＃xff0c;查看控制台结果

结果Sender为发送消息&＃xff0c;Receiver为接收消息。

五、发送对象消息

新建ObjectSender&＃xff0c;消息发送者&＃xff0c;如图

新建ObjectReceiver绑定队列queues &＃61; "object"监听器&＃xff0c;接收消息

新建Junit Test

运行之后查看控制台

接收对象

六、建立Topic交换机队列

建立TopicRabbitConfig.java&＃xff0c;建立队列

建立交换机

绑定交换机与队列并指定路由键

指定队列Sender&＃xff0c;匹配topic.#路由键,并指定交换机topicExchange

匹配topic.message路由键

匹配topic.#路由键

接收队列queues &＃61; "topic.message"

接收队列queues &＃61; "topic.messages"

建立Junit&＃xff0c;代码如下:

运行topic查看队列接收情况

由于绑定关系

因此是topic.messages这个队列接收到消息

回到Junit运行

运行结果

可以看到queues &＃61; "topic.message"和queues &＃61; "topic.messages"这2个队列都收到了消息。

因为绑定关系

而发送者

对于这2个路由键都符合

返回Junit运行

运行结果

因为发送者

topic.messages只匹配

这个路由键&＃xff0c;也就是

这个队列

七、Fanout广播

广播FanoutRabbitConfig代码如下:

import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.FanoutExchange;import org.springframework.amqp.core.Queue;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;&＃64;Configurationpublic class FanoutRabbitConfig { &＃64;Bean public Queue AMessage() { return new Queue("fanout.A"); } &＃64;Bean public Queue BMessage() { return new Queue("fanout.B"); } &＃64;Bean public Queue CMessage() { return new Queue("fanout.C"); } &＃64;Bean FanoutExchange fanoutExchange() { return new FanoutExchange("fanoutExchange"); } &＃64;Bean Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(AMessage).to(fanoutExchange); } &＃64;Bean Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(BMessage).to(fanoutExchange); } &＃64;Bean Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(CMessage).to(fanoutExchange); }

广播交换机fanoutExchange绑定了3个队列

队列1:

队列2&＃xff1a;

队列3&＃xff1a;

队列发送者&＃xff0c;广播交换机:

Junit工具如下

运行Junit

可以看到广播交换机的3个队列都接收到了消息

八、Many队列

队列监听多次

接收者1:

接收者2:

发送者1&＃xff1a;

发送者2&＃xff1a;

新建Junit&＃xff1a;

运行manyToMany

可以看到队列交错运行&＃xff0c;同一个队列对应关系Many To Many

以上就是分享的Spring Boot集成RabbitMQ实战&＃xff0c;有机会分享一下自己的消息中转推送中心系统。