专栏首页JackieZhengSpring Boot系列——7步集成RabbitMQ

Spring Boot系列——7步集成RabbitMQ

RabbitMQ是一种我们经常使用的消息中间件,通过RabbitMQ可以帮助我们实现异步、削峰的目的。

今天这篇,我们来看看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,usernamepassword四个属性即可

其他属性都有各自的含义,比如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实战——Profile

      看到Profile这个关键字,或许你从来没有正眼瞧过他,又或者脑海中有些模糊的印象,比如除了这里Springmvc中的Profile,maven中也有Pro...

    JackieZheng
  • Hadoop阅读笔记(一)——强大的MapReduce

    前言:来园子已经有8个月了,当初入园凭着满腔热血和一脑门子冲动,给自己起了个响亮的旗号“大数据 小世界”,顿时有了种世界都是我的,世界都在我手中的赶脚。可是.....

    JackieZheng
  • 把玩爬虫框架Gecco

    如果你现在接到一个任务,获取某某行业下的分类。 作为一个非该领域专家,没有深厚的运营经验功底,要提供一套摆的上台面且让人信服的行业分类,恐怕不那么简单。 找不到...

    JackieZheng
  • SpringBoot集成Swagger2

    在一些接口项目中,API的使用很频繁,所以一款API在线文档生成和测试工具非常有必要。而Swagger UI就是这么一款很实用的在线工具 本博客介绍如何在公司...

    SmileNicky
  • Java中使用线程时,请不要忘记Spring TaskExecutor组件

    当我们实现的web应用程序需要长时间运行一个任务时,Spring TaskExecutor管理组件是一个很好选择,会给我们代码的实现提供很大的方便,也会节省时间...

    用户1289394
  • rabbitMQ实现可靠消息投递 原

        RabbitMQ消息的可靠性主要包括两方面,一方面是通过实现消费的重试机制(通过@Retryable来实现重试,可以设置重试次数和重试频率,但是要保证幂...

    chinotan
  • Swagger2+springboot-mvc+hibernate-validator 可视化视图加参数校验

    我的博客即将同步至腾讯云+社区,邀请大家一同入驻:https://cloud.tencent.com/developer/support-plan?invite...

    用户5927264
  • 《SpringCloud篇:002Ribbon服务之间的负载均衡》

    Ribbon是为了实现服务与服务之间调用的负载均衡,是客户端负载均衡,并且默认采用轮询的策略。(类似于Nginx,但是Ribbon是用于服务端的,Nginx是客...

  • 05Ribbon负载均衡服务调用

    http://localhost/consumer/payment/get/130

    Remember_Ray
  • Spring Cloud Hystrix - 服务容错

    在微服务架构中,由于某个服务的不可用导致一系列的服务崩溃,被称之为雪崩效应。所以防御服务的雪崩效应是必不可少的,在Spring Cloud中防雪崩的利器就是Hy...

    端碗吹水

扫码关注云+社区

领取腾讯云代金券