0
0

[RabbitMQ] Spring Boot整合RabbitMQ

🌸个人主页:https://blog.csdn.net/2301_80050796?spm=1000.2115.3001.5343
🏵️热门专栏:
🧊 Java基本语法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12615970.html?spm=1001.2014.3001.5482
🍕 Collection与数据结构 (92平均质量分)https://blog.csdn.net/2301_80050796/category_12621348.html?spm=1001.2014.3001.5482
🧀线程与网络(96平均质量分) https://blog.csdn.net/2301_80050796/category_12643370.html?spm=1001.2014.3001.5482
🍭MySql数据库(93平均质量分)https://blog.csdn.net/2301_80050796/category_12629890.html?spm=1001.2014.3001.5482
🍬算法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12676091.html?spm=1001.2014.3001.5482
🍃 Spring(97平均质量分)https://blog.csdn.net/2301_80050796/category_12724152.html?spm=1001.2014.3001.5482
🎃Redis(97平均质量分)https://blog.csdn.net/2301_80050796/category_12777129.html?spm=1001.2014.3001.5482
🐰RabbitMQ(97平均质量分) https://blog.csdn.net/2301_80050796/category_12792900.html?spm=1001.2014.3001.5482
感谢点赞与关注~~~
在这里插入图片描述

目录

RabbitMQ开发,Spring也提供了一些便利.下面我们就来学习如何使用Spring操作RabbitMQ.

1. 工作队列模式

步骤:

  1. 引入依赖
  2. 编写yml设置,基本信息配置.
  3. 编写生产者代码.
  4. 编写消费者代码.(定义监听类,使用@RabbitListener注解完成队列监听).
  5. 观察运行结果.
  • 引入依赖 在我们创建Spring项目的依赖的时候,我们可以引入Spring For RabbitMQ依赖.在这里插入图片描述 也可以手动导入xml依赖:
<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency>
  • 添加配置 我们需要在yml文件中配置服务器的IP地址,RabbitMQ的服务端口号(默认5672),用户名,密码,以及我们要使用的虚拟机.
spring:application:name: rabbitmq-spring
  rabbitmq:host: 39.105.137.64
    port:5672username: jiangruijia
    password: qwe123524
    virtual-host: /
  • 编写生产者代码 为了方便测试,我们通过接口来发送信息. 首先我们需要在Constant类中定义队列的名称.
publicstaticfinalString WORK_QUEUE ="work_queue";

之后在config中声明队列.

@ConfigurationpublicclassRabbitMQConfig{@BeanpublicQueueworkQueue(){returnQueueBuilder.durable(Constant.WORK_QUEUE).build();//指定队列的名称}}

在

durable

中写入的是队列的名称,最后方法返回的是创建好的队列.

@RestController@RequestMapping("/producer")publicclassProducerController{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping("/work")publicStringwork(){
        rabbitTemplate.convertAndSend("",Constant.WORK_QUEUE,"发送消息");//指定发送消息的交换机和队列,以及发送的消息return"发送成功";}}

我们在编写生产者代码的时候,我们需要注入

RabbitTemplate

对象(类似于Redis中的StringRedisTemplate对象),使用这个对象来操作RabbitMQ.之后我们需要在一个方法中发送信息,

convertAndSend

创建了生产者并向指定的队列中发送了信息,在

convertAndSend

需要指定交换机,队列名称和需要发送的消息.
我们来运行代码,并访问指定的端口,从RabbitMQ的管理界面中查看消息.
在这里插入图片描述

在这里插入图片描述
我们可以从队列中获取到预期的消息:
在这里插入图片描述

  • 编写消费者代码
@ComponentpublicclassWorkListener{//工作模式中,消费者有两个@RabbitListener(queues =Constant.WORK_QUEUE)//监听指定的队列publicvoidworkListener1(Message message){System.out.println("workListener1收到消息:"+message);}@RabbitListener(queues =Constant.WORK_QUEUE)publicvoidworkListener2(Message message){System.out.println("workListener2收到消息:"+message);}}

在创建消费者的时候,我们需要使用

@Component

把消费者类注入到Spring中,之后,我们要在对应的方法上添加

@RabbitListener

,指定该方法为监听者,在后面的

queues

属性中指定需要监听的队列.
运行上述代码:在运行之后,我们发现消息队列中的信息减少了.而且在控制台中也打印出了收到的对应的消息.
在这里插入图片描述
在这里插入图片描述
在

@RabbitListener

修饰的方法的参数中,不仅仅可以指定Message类型的参数,还可以指定一下常见的参数:

  1. String: 返回消息的内容
  2. Message: 注意是org.springframework.amqp.core.Message包中的Message,在导包的时候不要导错.返回的是原始的消息体以及消息的属性,如消息ID,内容,队列.
  3. Channel: RabbitMQ中的通道消息对象,可以用于进行更高级的操作,比如手动确认.

2. Publish/Subscribe(发布订阅模式)

在发布/订阅模式中,多了一个交换机的角色.exchange常见的有三种类型,我们在前面介绍过,发别是:fanout广播模式,将消息交给所有绑定交换机的队列.direct定向模式,把消息交给符合指定RoutingKey的队列.topic通配符模式,把消息交给符合routing pattern(路由模式)的队列,只不过这里的RoutingKey中含有通配符.而定向模式中的RoutingKey是写死的.

  • 编写生产者代码 和简单模式,工作队列模式最大的区别就是需要创建交换机,并绑定交换机和队列. 首先我们声明队列,交换机的名称.
publicstaticfinalString FANOUT_QUEUE1 ="fanout_queue1";publicstaticfinalString FANOUT_QUEUE2 ="fanout_queue2";publicstaticfinalString FANOUT_EXCHANGE ="fanout_exchange";

之后我们需要在声明交换机和队列,并绑定交换机和队列.之后使用

@Bean

注解进行注入.

@BeanpublicQueuefanoutQueue1(){returnQueueBuilder.durable(Constant.FANOUT_QUEUE1).build();}@BeanpublicQueuefanoutQueue2(){returnQueueBuilder.durable(Constant.FANOUT_QUEUE2).build();}@BeanpublicFanoutExchangefanoutExchange(){returnExchangeBuilder.fanoutExchange(Constant.FANOUT_EXCHANGE).durable(true).build();}

在创建交换机的时候,我们需要使用

 ExchangeBuilder.fanoutExchange

方法来指定交换机的名字,之后使用

durable

指定交换机为持久化.注意在返回交换机的时候,必须返回的是具体类型的交换机.队列的创建方法和前面的创建方法相同.
之后我们对交换机和队列进行绑定.

@BeanpublicBindingbinding1(@Qualifier("fanoutExchange")FanoutExchange exchange,@Qualifier("fanoutQueue1")Queue queue){returnBindingBuilder.bind(queue).to(exchange);}@BeanpublicBindingbinging2(@Qualifier("fanoutExchange")FanoutExchange exchange,@Qualifier("fanoutQueue2")Queue queue){returnBindingBuilder.bind(queue).to(exchange);}

在绑定的时候,我们需要使用

@Qualifier

注解进行注入,注意注入的交换机类型必须和Spring容器中的交换机类型一致,都是fanout类型的交换机.

bind

方法中写入的是需要绑定的队列,

to

中写入的是需要绑定的交换机.
之后我们使用接口发送消息.

@RequestMapping("/fanout")publicStringfanout(){
    rabbitTemplate.convertAndSend(Constant.FANOUT_EXCHANGE,"","fanout交换机发送消息");//指定发送的交换机即可,由于是广播模式,RoutingKey不需要指定return"发送成功";}
  • 编写消费者代码 定义监听类,处理收到的消息即可.
@ComponentpublicclassFanoutListener{@RabbitListener(queues =Constant.FANOUT_QUEUE1)publicvoidListenerQueue1(String message){System.out.println("queue1收到消息:"+message);}@RabbitListener(queues =Constant.FANOUT_QUEUE2)publicvoidListenerQueue2(String message){System.out.println("queue2收到消息:"+message);}}

运行项目,之后访问Controller对应的接口.
在这里插入图片描述
在这里插入图片描述
rabbbitMQ的管理界面,消息条数发生了波动,控制台成功收到了生产者发送的消息.

3. Routing(路由模式)

交换机类型为direct类型的时候,消息会把消息交给符合指定RoutingKey的队列.队列和交换机的绑定,不是任意绑定了,而是需要指定一个RoutingKey.交换机也不再把消息交给每一个绑定的key,而是根据消息的RoutingKey进行判断,只有队列的RoutingKey和消息的RoutingKey完全一致,才会收到消息.

  • 编写生产者代码 指定队列和交换机的名称
publicstaticfinalString DIRECT_EXCHANGE ="direct_exchange";publicstaticfinalString DIRECT_QUEUE1 ="direct_queue1";publicstaticfinalStringDIREct_QUEUE2="direct_queue2";

接下来声明交换机和队列

@BeanpublicQueuedirectQueue1(){returnQueueBuilder.durable(Constant.DIRECT_QUEUE1).build();}@BeanpublicQueuedirectQueue2(){returnQueueBuilder.durable(Constant.DIREct_QUEUE2).build();}@BeanpublicDirectExchangedirectExchange(){returnExchangeBuilder.directExchange(Constant.DIRECT_EXCHANGE).durable(true).build();}

接下来绑定队列和交换机,指定绑定时候的BindingKey.

@BeanpublicBindingdirectBinding1(@Qualifier("directExchange")DirectExchange exchange,@Qualifier("directQueue1")Queue queue){returnBindingBuilder.bind(queue).to(exchange).with("orange");}@BeanpublicBindingdirectBinding2(@Qualifier("directExchange")DirectExchange exchange,@Qualifier("directQueue2")Queue queue){returnBindingBuilder.bind(queue).to(exchange).with("green");}@BeanpublicBindingdirectBinding3(@Qualifier("directExchange")DirectExchange exchange,@Qualifier("directQueue2")Queue queue){returnBindingBuilder.bind(queue).to(exchange).with("red");}

相比广播模式,多了

with

指定了绑定时候的BindingKey.
使用接口发送消息.

@RequestMapping("/direct")publicStringdirect(String routingKey){
    rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE,routingKey,"direct生产者发送消息");return"发送成功";}

与前面广播不同的是,在发送消息的时候,需要在参数中指定RoutingKey,在给交换机发送消息的时候,需要把传入的RoutingKey传入

convertAndSend

中.

  • 编写消费者代码 消费者的编写方式和前面几种方式都差不多,都是在指定的方法上添加@RabbitListener注解,在注解中指定需要监听的队列.
@ComponentpublicclassDirectListener{@RabbitListener(queues =Constant.DIRECT_QUEUE1)publicvoidListenerQueue1(String message){System.out.println("监听队列1:"+message);}@RabbitListener(queues =Constant.DIREct_QUEUE2)publicvoidListenerQueue2(String message){System.out.println("监听队列2:"+message);}}

运行项目并访问指定接口.

http://127.0.0.1:8080/producer/direct?routingKey=green

在这里插入图片描述

http://127.0.0.1:8080/producer/direct?routingKey=red

在这里插入图片描述

http://127.0.0.1:8080/producer/direct?routingKey=orange

在这里插入图片描述
在这里插入图片描述

4. Topics(通配符模式)

Topic和Routing模式的区别是:

  1. topics模式使用的交换机类型为topic(Routing模式使用的交换机类型为direct)
  2. topic类型的交换机在匹配规则上进行了扩展,BindingKey支持通配符匹配.
  • 编写生产者代码 首先还是指定队列和交换机的名称
publicstaticfinalString TOPICS_QUEUE1 ="topics_queue1";publicstaticfinalString TOPICS_QUEUE2 ="topics_queue2";publicstaticfinalString TOPICS_EXCHANGE ="topics_exchange";

之后声明交换机和队列.

@BeanpublicQueuetopicsQueue1(){returnQueueBuilder.durable(Constant.TOPICS_QUEUE1).build();}@BeanpublicQueuetopicsQueue2(){returnQueueBuilder.durable(Constant.TOPICS_QUEUE2).build();}@BeanpublicTopicExchangetopicExchange(){returnExchangeBuilder.topicExchange(Constant.TOPICS_EXCHANGE).durable(true).build();}

绑定交换机和队列,并指定BindingKey

@BeanpublicBindingtopicsBinding1(@Qualifier("topicsExchange")TopicExchange exchange,@Qualifier("topicsQueue1")Queue queue){returnBindingBuilder.bind(queue).to(exchange).with("#.error");}@BeanpublicBindingtopicsBinding2(@Qualifier("topicsExchange")TopicExchange exchange,@Qualifier("topicsQueue2")Queue queue){returnBindingBuilder.bind(queue).to(exchange).with("#.info");}

之后我们使用接口来接收信息.

@RequestMapping("/topics")publicStringtopics(String routingKey){
    rabbitTemplate.convertAndSend(Constant.TOPICS_EXCHANGE,routingKey,"topics生产者发送消息");return"发送成功";}
  • 编写消费者代码
@ComponentpublicclassTopicsListener{@RabbitListener(queues =Constant.TOPICS_QUEUE1)publicvoidtopicsListener1(String message){System.out.println("topics队列1接收到消息"+message);}@RabbitListener(queues =Constant.TOPICS_QUEUE2)publicvoidtopicsListener2(String message){System.out.println("topics队列2接收到消息"+message);}}

运行项目,访问接口,传递指定参数,观察结果

http://127.0.0.1:8080/producer/topics?routingKey=404.error

在这里插入图片描述

http://127.0.0.1:8080/producer/topics?routingKey=200.info

在这里插入图片描述
在这里插入图片描述

5. 实际案例

作为一个消息队列,RabbitMQ也可以用作应用程序之间的通信.上述代码和生产者和消费者代码放在不同的项目模块中即可完成不同的应用程序通信.
比如我们需要实现下面的功能:
用户下单成功之后,通知物流系统进行发货.
在这里插入图片描述
这里我们只做应用通信,我们不对业务逻辑做具体实现.

5.1 创建项目

我们在创建项目的时候,把两个项目放在同一个空项目中.

  1. 创建一个空项目在这里插入图片描述
  2. 之后在这个项目中创建一个模块在这里插入图片描述
  3. 后续流程和创建SpringBoot项目一样.在这里插入图片描述 在模块创建的时候,我们在每个项目中都会引入如下的依赖:在这里插入图片描述 最终的项目结构如下:在这里插入图片描述我们在创建项目之后有可能无法加载项目为Maven项目,我们可以在指定的项目的pom文件中右键—>添加为Maven项目即可.

5.2 订单系统(生产者)

  1. 完善配置信息
spring:application:name: logistics-service
  rabbitmq:host: 39.105.137.64
    port:5672username: jiangruijia
    password: qwe123524
    virtual-host: /
  1. 声明队列
@ConfigurationpublicclassRabbitMQConfig{@BeanpublicQueueworkQueue(){returnQueueBuilder.durable("order.create").build();}}
  1. 编写下单接口,下单成功之后,发送订单消息到队列.
@RestController@RequestMapping("/order")publicclassOrderController{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping("/createOrder")publicStringcreateOrder(){String order = UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend("","order.create","下单成功:"+order);return"下单成功";}}
  1. 启动服务,访问对应接口,观察结果 消息条数出现了波动.在这里插入图片描述 从队列中查看消息:在这里插入图片描述

5.3 物流系统(消费者)

  1. 完善配置信息 8080端口号已经订单系统被占领,所以我们需要更改物流系统的端口号为9090.
spring:application:name: logistics-service
  rabbitmq:host: 39.105.137.64
    port:5672username: jiangruijia
    password: qwe123524
    virtual-host: /
server:port:9090
  1. 消费者监听队列
@ComponentpublicclassOrderControllerListener{@RabbitListener(queues ="order.create")publicvoidListenerQueue(String message){System.out.println("接收到订单:"+message);//此处业务逻辑省略}}

启动服务,观察结果:
在这里插入图片描述
在这里插入图片描述
我们看到了消费者成功接收到了订单id.

标签： java-rabbitmq rabbitmq spring boot
本文转载自: https://blog.csdn.net/2301_80050796/article/details/142673534
版权归原作者 LileSily 所有， 如有侵权，请联系我们删除。
登录后发布评论

“[RabbitMQ] Spring Boot整合RabbitMQ”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

docker无法搜索和拉取镜像

visual studio之安装详解

【Linux】版本控制器git、调试器gdb/cgdb

面试题：RabbitMQ（一）

Vllm进行Qwen2-vl部署（包含单卡多卡部署及爬虫请求）

MySQL wait_timeout参数介绍

Python毕业设计选题：基于django+vue的企业it资产管理系统

文章导航