脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → Spring Boot集成RabbitMQ

Spring Boot系列教程之7步集成RabbitMQ的方法

 更新时间：2018年11月12日 11:44:59   作者：JackieZheng  
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。下面这篇文章主要给大家介绍了关于Spring Boot之7步集成RabbitMQ的相关资料，需要的朋友可以参考下

前言

RabbitMQ是一种我们经常使用的消息中间件，RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

今天这篇，我们来看看Spring Boot是如何集成RabbitMQ，发送消息和消费消息的。同时我们介绍下死信队列。

集成RabbitMQ

集成RabbitMQ只需要如下几步即可

1、添加maven依赖

<!--rabbitmq-->

<dependency>

 <groupId>org.springframework.boot</groupId>

 <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

2、添加配置文件application.yaml

在application.yaml添加配置内容如下

spring: rabbitmq:
 host: 192.168.1.161
 port: 5672
 username: guest
 password: guest
 cache:
 channel: size: 10
 listener:
 type: simple
 simple:
 acknowledge-mode: auto
 concurrency: 5
 default-requeue-rejected: true
 max-concurrency: 100
 retry:
 enabled: true # initial-interval: 1000ms
 max-attempts: 3 # max-interval: 1000ms
 multiplier: 1
 stateless: true # publisher-confirms: true</pre>

注意：

这里最基本的配置只需要配置host,port,username和password四个属性即可

其他属性都有各自的含义，比如retry是用于配置重试策略的，acknowledge-mode是配置消息接收确认机制的。

3、编写配置类

编写RabbitConfig配置类，采用Java Configuration的方式配置RabbitTemplate、Exchange和Queue等信息，具体如下所示

package com.jackie.springbootdemo.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.HashMap;
import java.util.Map;

@Configuration public class RabbitMQConfig implements InitializingBean { @Autowired
 SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory;

 @Override
 public void afterPropertiesSet() throws Exception {
 simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter());
 } @Bean("jackson2JsonMessageConverter")
 public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ConnectionFactory connectionFactory) {
 return new Jackson2JsonMessageConverter();
 } @Bean("rabbitTemplate")
 @Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
 public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
  @Qualifier("jackson2JsonMessageConverter") Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
 RabbitTemplate template = new RabbitTemplate(connectionFactory);
 template.setMessageConverter(new Jackson2JsonMessageConverter());
 return template;
 } // --------------------- 声明队列 ------------------------
 @Bean
 public Queue demoQueue() {
 return new Queue("demo_queue");
 } // --------------------- 声明exchange ------------------------ @Bean
 public DirectExchange demoExchange() {
 return new DirectExchange("demo_exchange");
 } // --------------------- 队列绑定 ------------------------
 @Bean
 public Binding bindingAlbumItemCreatedQueue(DirectExchange demoExchange,
  Queue demoQueue) {
 return BindingBuilder.bind(demoQueue).to(demoExchange).with("100");
 } }

注意

这里声明了Direct模式的Exchange，声明一个Queue，并通过routing-key为100将demo_queue绑定到demo_exchange，这样demo_queue就可以接收到demo_exchange发送的消息了。

4、编写消息发送类

package com.jackie.springbootdemo.message;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component public class Sender implements RabbitTemplate.ConfirmCallback { private RabbitTemplate rabbitTemplate;

 /**
 * 构造方法注入 */ @Autowired
 public Sender(RabbitTemplate rabbitTemplate) {
 this.rabbitTemplate = rabbitTemplate;
 rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话，那回调就是最后设置的内容
 } public void sendMsg(String content) {
 rabbitTemplate.convertAndSend("demo_exchange", "100", content);
 } /**
 * 回调 */ @Override
 public void confirm(CorrelationData correlationData, boolean ack, String cause) {
 System.out.println(" 回调id:" + correlationData);
 if (ack) {
 System.out.println("消息成功消费");
 } else {
 System.out.println("消息消费失败:" + cause);
 }
 } }

注意

发送内容content，路由到routing-key为100上，则我们就可以在demo_queue队列中看到发送的消息内容了

confirm函数是回调函数，这里因为没有消费者，且acknoledge-mode是auto(其他两种值分别是none和manual)，所以ack是false。

5、编写发送消息测试类

package com.jackie.springbootdemo;

import com.jackie.springbootdemo.message.Sender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.web.WebAppConfiguration;

@RunWith(SpringRunner.class) @SpringBootTest(classes = SpringbootDemoApplication.class) @WebAppConfiguration public class RabbitApplicationTests { @Autowired
 Sender sender;

 @Test
 public void contextLoads() throws Exception {
 sender.sendMsg("test");
 } }

运行该测试类，我们可以看到如下结果

6、编写消息消费类

package com.jackie.springbootdemo.message;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component public class Receiver { @RabbitListener(queues = "demo_queue")
 public void created(String message) {
 System.out.println("orignal message: " + message);
 } }

注意

消息消费类也非常简单，添加注解@RabbitListener，指定要监听的队列名称即可

除了注解@RabbitListener，我们经常还能看到@RabbitHandler，这两个注解可以配合起来使用。

@RabbitListener 标注在类上面表示当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，根据 MessageConverter 转换后的参数类型，形如

@RabbitListener(queues = "demo_queue") public class Receiver { @RabbitHandler public void processMessage1(String message) {
 System.out.println(message);
 } @RabbitHandler
 public void processMessage2(byte[] message) {
 System.out.println(new String(message));
 } }

7、运行消息发送测试类

从执行结果可以看到，因为有了消费者，所以这次打印的结果是"消息消费成功"

而且，我们看到Receiver类将消息消费并打印出消息的内容为"test"。

代码已经提交至项目rome：https://github.com/DMinerJackie/rome （本地下载）

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对脚本之家的支持。

您可能感兴趣的文章:

相关文章

  • java留言管理系统中模糊查询实例分享

    java留言管理系统中模糊查询实例分享

    这篇文章主要为大家详细介绍了基于MVC+DAO的留言管理系统中java模糊查询的简单使用方法，感兴趣的小伙伴们可以参考一下
    2016-04-04
  • java使用观察者模式异步短信/邮箱提醒用户群

    java使用观察者模式异步短信/邮箱提醒用户群

    这篇文章主要为大家详细介绍了java使用观察者模式异步短信和邮箱提醒用户群，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2018-11-11
  • JDK 8和JDK 17的区别和新特性大全

    JDK 8和JDK 17的区别和新特性大全

    这篇文章主要给大家介绍了关于JDK 8和JDK 17的区别和新特性的相关资料,文中总结一些Jdk8到Jdk17的一些新特性,给大家选择jdk版本的时候有些参考性,需要的朋友可以参考下
    2023-06-06
  • 从零开始Java实现Parser Combinator

    从零开始Java实现Parser Combinator

    这篇文章主要为大家介绍了从零开始Java实现Parser Combinator过程及原理详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
    2023-05-05
  • Java编写Mapreduce程序过程浅析

    Java编写Mapreduce程序过程浅析

    MapReduce是一种用于处理大规模数据集的并行编程模型，其特点高效性和可扩展性，在本文中，我们将深入了解MapReduce，并使用Java编写一个简单的MapReduce程序,需要的朋友可以参考下
    2023-05-05
  • Java线程池详细解读

    Java线程池详细解读

    这篇文章主要给大家介绍了关于Java中方法使用的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2021-08-08
  • Java算法之BFS,DFS,动态规划和贪心算法的实现

    Java算法之BFS,DFS,动态规划和贪心算法的实现

    广度优先搜索（BFS）和深度优先搜索（DFS）是图遍历算法中最常见的两种算法，主要用于解决搜索和遍历问题。动态规划和贪心算法则用来解决优化问题。本文就来看看这些算法的具体实现吧
    2023-04-04
  • SpringBoot整合OpenFeign的坑

    SpringBoot整合OpenFeign的坑

    最近试用SpringBoot+K8S，遇到了个坑，通过OpenFeign请求返回值LocalDateTime发生了异常，本文就详细的介绍一下解决方法，感兴趣的可以了解一下
    2021-07-07
  • java非官方常用类MessageInfo消息接口示例

    java非官方常用类MessageInfo消息接口示例

    这篇文章主要为大家介绍了java非官方常用类MessageInfo消息接口使用示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-09-09
  • SSM项目使用拦截器实现登录验证功能

    SSM项目使用拦截器实现登录验证功能

    这篇文章主要介绍了在SSM项目中如何使用拦截器，实现登录验证功能。文中的实现步骤讲解详细，感兴趣的小伙伴可以了解一下
    2022-01-01

最新评论