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 (本地下载)

总结

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

(0)

相关推荐

  • spring boot整合RabbitMQ实例详解(Fanout模式)

    1.Fanout Exchange介绍 Fanout Exchange 消息广播的模式,不管路由键或者是路由模式,会把消息发给绑定给它的全部队列,如果配置了routing_key会被忽略. 如上图所示,即当使用fanout交换器时,他会将消息广播到与该交换器绑定的所有队列上,这有利于你对单条消息做不同的反应. 例如存在以下场景:一个web服务要在用户完善信息时,获得积分奖励,这样你就可以创建两个对列,一个用来处理用户信息的请求,另一个对列获取这条消息是来完成积分奖励的任务. 2.代码示例 1).

  • Spring Boot与RabbitMQ结合实现延迟队列的示例

    背景 何为延迟队列? 顾名思义,延迟队列就是进入该队列的消息会被延迟消费的队列.而一般的队列,消息一旦入队了之后就会被消费者马上消费. 场景一:在订单系统中,一个用户下单之后通常有30分钟的时间进行支付,如果30分钟之内没有支付成功,那么这个订单将进行一场处理.这是就可以使用延时队列将订单信息发送到延时队列. 场景二:用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作.这时候就可以将用户指令发送到延时队列,当指令设定的时间到了再将指令推送到只能设备. 延迟队列能做什么? 延迟队列多用于需

  • spring boot集成rabbitmq的实例教程

    一.RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

  • Spring Boot整合RabbitMQ实例(Topic模式)

    1.Topic交换器介绍 Topic Exchange 转发消息主要是根据通配符. 在这种交换机下,队列和交换机的绑定会定义一种路由模式,那么,通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息. 在这种交换机模式下: 路由键必须是一串字符,用句号(.) 隔开,比如说 agreements.us,或者 agreements.eu.stockholm 等. 路由模式必须包含一个 星号(*),主要用于匹配路由键指定位置的一个单词,比如说,一个路由模式是这样子:agreements..b.*

  • Spring boot集成RabbitMQ的示例代码

    RabbitMQ简介 RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统 MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法.应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们.消息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术.排队指的是应用程序通过 队列来通信.队列的使用除去了接收和发送应用程序同时执行的要求. AMQP就是一个协议

  • 详解Spring Boot 配置多个RabbitMQ

    闲话 好久没有写博客了,6月份毕业,因为工作原因,公司上网受限,一直没能把学到的知识点写下来,工作了半年,其实学到的东西也不少,但是现在回忆起来的东西少之又少,有时甚至能在同个问题中踩了几次,越来越觉得及时记录一下学到的东西很重要. 好了,闲话少说,写下这段时间学习的东西,先记录一下用spring Boot配置多个RabbitMQ的情况... 最近公司新启动一个新平台的项目,需要用微服务这个这几年很火的概念来做,所以就学习了Spring Boot方面的知识,给同事展示Spring Boot的一些

  • springboot集成rabbitMQ之对象传输的方法

    rabbitMQ的安装方法网上有很多教程,这里就不重复了. 在springboot上使用rabbitMQ传输字符串和对象,本文所给出的例子是在两个不同的项目之间进行对象和和字符串的传输. rabbitMQ的依赖(在两个项目中一样的配置): <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId>

  • 详解spring boot集成RabbitMQ

    RabbitMQ作为AMQP的代表性产品,在项目中大量使用.结合现在主流的spring boot,极大简化了开发过程中所涉及到的消息通信问题. 首先正确的安装RabbitMQ及运行正常. RabbitMQ需啊erlang环境,所以首先安装对应版本的erlang,可在RabbitMQ官网下载 # rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm 使用yum安装RabbitMQ,避免缺少依赖包引起的安装失败 # yum install rabbitmq-s

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

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

  • Spring Boot高级教程之Spring Boot连接MySql数据库

    Spring Boot可以大大简化持久化任务,几乎不需要写SQL语句,在之前章节"Spring Boot 构建框架"中我们新建了一个Spring Boot应用程序,本章在原有的工程中与数据库建立连接. Spring Boot有两种方法与数据库建立连接,一种是使用JdbcTemplate,另一种集成Mybatis,下面分别为大家介绍一下如何集成和使用这两种方式. 1. 使用JdbcTemplate <dependency> <groupId>mysql</g

  • spring boot实战教程之shiro session过期时间详解

    前言 众所周知在spring boot内,设置session过期时间只需在application.properties内添加server.session.timeout配置即可.在整合shiro时发现,server.session.timeout设置为7200,但未到2小时就需要重新登录,后来发现是shiro的session已经过期了,shiro的session过期时间并不和server.session.timeout一致,目前是采用filter的方式来进行设置. ShiroSessionFil

  • Spring温故而知新系列教程之AOP代理

    AOP的概念 AOP:Aspect-Oriented Programming(面向切面编程),维基百科的解释如下:Aspect是一种新的模块化机制,用来描述分散在对象.类或者函数中的横切关注点,从关注点中分离出横切关注点是面向切面的程序设计的核心概念.分离关注点使解决特定领域问题的代码从业务逻辑中独立出来,业务逻辑的代码中不在含有针对特定领域问题的代码的调用,业务逻辑同特定领域问题的关系通过切面来封装.维护,这样原本分散在整个应用程序中的变动就可以很好地管理起来.从AOP的角度,应用可以分为横切

  • SpringBoot系列教程之dubbo和Zookeeper集成方法

    今日学习新的内容:dubbo   dubbo是阿里巴巴公司开源的一个高性能优秀的服务框架,一款高性能.轻量级的开源Java RPC框架,它提供了三大核心能力:面向接口的远程方法调用,智能容错和负载均衡,以及服务自动注册和发现. zookeeper   zooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,它是一个为分布式应用提供一致性服务的软件,提供的功能包括:配置维护.域名服务.分布式同步.组服务等. SpringBoot整合dubbo和Zookeeper 1.了解Dubbo基本

  • Spring Boot系列教程之死信队列详解

    前言 在说死信队列之前,我们先介绍下为什么需要用死信队列. 如果想直接了解死信对接,直接跳入下文的"死信队列"部分即可. ack机制和requeue-rejected属性 我们还是基于上篇<Spring Boot系列--7步集成RabbitMQ>的demo代码来说. 在项目springboot-demo我们看到application.yaml文件部分配置内容如下 ... listener: type: simple simple: acknowledge-mode: aut

  • java JDBC系列教程之JDBC类的简析与JDBC的基础操作

    什么是JDBC? 概念:JAVA Database Connectivity Javas数据库连接,Java语言操作数据库接口,然后由各个数据库厂商去实现这个接口,提供数据库驱动java包,我们可以使用这套接口,真正执行的是jar驱动包中的实习类 使用一张图让大家更为直观的理解: coder就是写这套接口的程序员 JDBC的使用步骤 1.导入驱动jar包 2.注册驱动 3.获取数据库连接对象 4.定义sql执行语句 5.获取sql语句执行对象 6.执行sql语句返回结果 7.处理结果 8.释放结

  • Spring Boot 自定义 Shiro 过滤器无法使用 @Autowired问题及解决方法

    在 Spring Boot 中集成 Shiro,并使用 JWT 进行接口认证. 为了统一对 Token 进行过滤,所以自定义了一个 JwtTokenFilter 过滤器. 期间遇到了以下几个问题,这里逐一进行记录,以备日后查阅. 问题一:JwtTokenFilter 无法使用 @Autowired 因为自定义了一个 JWT Token 工具类,用来解析和创建 Token,JwtTokenFilter 中需要用到此工具类,这里本来可以直接手动进行 new 一个新的实例,但由于在 Spring 配置

  • Spring Boot Maven 打包可执行Jar文件的实现方法

    Maven pom.xml 必须包含 <packaging>jar</packaging> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <configuration> <fork>

  • Spring Boot 与 Kotlin 使用JdbcTemplate连接MySQL数据库的方法

    之前介绍了一些Web层的例子,包括构建RESTful API.使用Thymeleaf模板引擎渲染Web视图,但是这些内容还不足以构建一个动态的应用.通常我们做App也好,做Web应用也好,都需要内容,而内容通常存储于各种类型的数据库,服务端在接收到访问请求之后需要访问数据库获取并处理成展现给用户使用的数据形式. 本文介绍在Spring Boot基础下配置数据源和通过 JdbcTemplate 编写数据访问的示例. 数据源配置 在我们访问数据库的时候,需要先配置一个数据源,下面分别介绍一下几种不同

随机推荐