详解spring boot集成RabbitMQ

 更新时间:2017年03月30日 10:11:15   作者:SamHxm   我要评论
RabbitMQ作为AMQP的代表性产品,在项目中大量使用。结合现在主流的spring boot,极大简化了开发过程中所涉及到的消息通信问题。

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-server-3.6.6-1.el7.noarch.rpm

启动RabbitMQ

# /sbin/service rabbitmq-server start

由于RabbitMQ默认提供的guest用户只能本地访问,所以额外创建用户用于测试

# /sbin/rabbitmqctl add_user test test123
用户名:test,密码:test123

开启web管理插件

# rabbitmq-plugins enable rabbitmq_management

并使用之前创建的用户登录,并设置该用户为administrator,虚拟主机地址为/

spring boot 引入相关依赖

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
</dependencies>

消息生产者

application.properties添加一下配置

spring.rabbitmq.host=192.168.1.107
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test123
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

spring boot配置类,作用为指定队列,交换器类型及绑定操作

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

  //声明队列
  @Bean
  public Queue queue1() {
    return new Queue("hello.queue1", true); // true表示持久化该队列
  }

  @Bean
  public Queue queue2() {
    return new Queue("hello.queue2", true);
  }

  //声明交互器
  @Bean
  TopicExchange topicExchange() {
    return new TopicExchange("topicExchange");
  }

  //绑定
  @Bean
  public Binding binding1() {
    return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
  }

  @Bean
  public Binding binding2() {
    return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
  }

}

共声明了2个队列,分别是hello.queue1,hello.queue2,交换器类型为TopicExchange,并与hello.queue1,hello.queue2队列分别绑定。

生产者类

import java.util.UUID;

import javax.annotation.PostConstruct;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
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, ReturnCallback {

  @Autowired
  private RabbitTemplate rabbitTemplate;

  @PostConstruct
  public void init() {
    rabbitTemplate.setConfirmCallback(this);
    rabbitTemplate.setReturnCallback(this);
  }

  @Override
  public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    if (ack) { 
      System.out.println("消息发送成功:" + correlationData); 
    } else { 
      System.out.println("消息发送失败:" + cause); 
    } 

  }

  @Override
  public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
    System.out.println(message.getMessageProperties().getCorrelationIdString() + " 发送失败");

  }

  //发送消息,不需要实现任何接口,供外部调用。
  public void send(String msg){

    CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

    System.out.println("开始发送消息 : " + msg.toLowerCase());
    String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
    System.out.println("结束发送消息 : " + msg.toLowerCase());
    System.out.println("消费者响应 : " + response + " 消息处理完成");
  }
}

要点:

1.注入RabbitTemplate

2.实现RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口(非必须)。
ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。ReturnCallback接口用于实现消息发送到RabbitMQ交换器,但无相应队列与交换器绑定时的回调。

3.实现消息发送方法。调用rabbitTemplate相应的方法即可。rabbitTemplate常用发送方法有

rabbitTemplate.send(message);  //发消息,参数类型为org.springframework.amqp.core.Message
rabbitTemplate.convertAndSend(object); //转换并发送消息。 将参数对象转换为org.springframework.amqp.core.Message后发送
rabbitTemplate.convertSendAndReceive(message) //转换并发送消息,且等待消息者返回响应消息。

针对业务场景选择合适的消息发送方式即可。

消息消费者

application.properties添加一下配置

spring.rabbitmq.host=192.168.1.107
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test123

spring.rabbitmq.listener.concurrency=2  //最小消息监听线程数
spring.rabbitmq.listener.max-concurrency=2 //最大消息监听线程数

消费者类

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

@Component
public class Receiver {

  @RabbitListener(queues = "hello.queue1")
  public String processMessage1(String msg) {
    System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue1队列的消息:" + msg);
    return msg.toUpperCase();
  }

  @RabbitListener(queues = "hello.queue2")
  public void processMessage2(String msg) {
    System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue2队列的消息:" + msg);
  }
}

由于定义了2个队列,所以分别定义不同的监听器监听不同的队列。由于最小消息监听线程数和最大消息监听线程数都是2,所以每个监听器各有2个线程实现监听功能。

要点:

1.监听器参数类型与消息实际类型匹配。在生产者中发送的消息实际类型是String,所以这里监听器参数类型也是String。

2.如果监听器需要有响应返回给生产者,直接在监听方法中return即可。

运行测试

import java.util.Date;

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.SpringJUnit4ClassRunner;

import com.sam.demo.rabbitmq.Application;
import com.sam.demo.rabbitmq.sender.Sender;

@RunWith(value=SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = Application.class)
public class RabbitTests {

  @Autowired
  private Sender sender;

  @Test
  public void sendTest() throws Exception {
    while(true){
      String msg = new Date().toString();
      sender.send(msg);
      Thread.sleep(1000);
    }
  }
}

输出:

开始发送消息 : wed mar 29 23:20:52 cst 2017
SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息:Wed Mar 29 23:20:52 CST 2017
SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息:Wed Mar 29 23:20:52 CST 2017
结束发送消息 : wed mar 29 23:20:52 cst 2017
消费者响应 : WED MAR 29 23:20:52 CST 2017 消息处理完成
------------------------------------------------
消息发送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886]
开始发送消息 : wed mar 29 23:20:53 cst 2017
SimpleAsyncTaskExecutor-1 接收到来自hello.queue1队列的消息:Wed Mar 29 23:20:53 CST 2017
SimpleAsyncTaskExecutor-2 接收到来自hello.queue2队列的消息:Wed Mar 29 23:20:53 CST 2017
结束发送消息 : wed mar 29 23:20:53 cst 2017
消费者响应 : WED MAR 29 23:20:53 CST 2017 消息处理完成
------------------------------------------------
消息发送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34]
开始发送消息 : wed mar 29 23:20:54 cst 2017
SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息:Wed Mar 29 23:20:54 CST 2017
SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息:Wed Mar 29 23:20:54 CST 2017
结束发送消息 : wed mar 29 23:20:54 cst 2017
消费者响应 : WED MAR 29 23:20:54 CST 2017 消息处理完成
------------------------------------------------

如果需要使用的其他的交换器类型,spring中都已提供实现,所有的交换器均实现org.springframework.amqp.core.AbstractExchange接口。

常用交换器类型如下:

Direct(DirectExchange):direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key完全匹配时, 才会被交换器投送到绑定的队列中去。

Topic(TopicExchange):按规则转发消息(最灵活)。

Headers(HeadersExchange):设置header attribute参数类型的交换机。

Fanout(FanoutExchange):转发消息到所有绑定队列。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • Java实现单例模式之饿汉式、懒汉式、枚举式

    Java实现单例模式之饿汉式、懒汉式、枚举式

    本篇文章主要介绍了Java实现单例的3种普遍的模式,饿汉式、懒汉式、枚举式。具有一定的参考价值,感兴趣的小伙伴们可以参考一下。
    2016-10-10
  • Spring Cloud Config Client超时及重试示例详解

    Spring Cloud Config Client超时及重试示例详解

    这篇文章主要给大家介绍了关于Spring Cloud Config Client超时及重试的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面来一起学习学习吧
    2018-05-05
  • 详解Spring AOP 拦截器的基本实现

    详解Spring AOP 拦截器的基本实现

    本篇文章主要介绍了详解Spring AOP 拦截器的基本实现,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-03-03
  • SpringBoot 使用Mybatis分页插件实现详解

    SpringBoot 使用Mybatis分页插件实现详解

    这篇文章主要介绍了SpringBoot 使用Mybatis分页插件实现详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-10-10
  • Java 高并发三:Java内存模型和线程安全详解

    Java 高并发三:Java内存模型和线程安全详解

    本文主要介绍Java高并发内存模型和线程安全的资料,这里整理详细的资料及1.原子性 2.有序性 3.可见性 4.Happen-Before 5.线程安全的概念,有需要的小伙伴可以参考下
    2016-09-09
  • java selenium教程环境搭建基于Maven

    java selenium教程环境搭建基于Maven

    本文主要介绍Java selenium 环境的安装,这里介绍了基于Maven的环境搭建,有需要的小伙伴可以参考下
    2016-08-08
  • Spring MVC 文件上传下载的实例

    Spring MVC 文件上传下载的实例

    本篇文章主要介绍了Spring MVC 文件上传下载的实例,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-01-01
  • java实现截取PDF指定页并进行图片格式转换功能

    java实现截取PDF指定页并进行图片格式转换功能

    这篇文章主要介绍了java实现截取PDF指定页并进行图片格式转换功能,本文通过实例代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下
    2019-09-09
  • 浅谈java多态的实现主要体现在哪些方面

    浅谈java多态的实现主要体现在哪些方面

    下面小编就为大家带来一篇浅谈java多态的实现主要体现在哪些方面。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2016-09-09
  • Intellij idea远程debug连接tomcat实现单步调试

    Intellij idea远程debug连接tomcat实现单步调试

    这篇文章主要介绍了Intellij idea远程debug连接tomcat实现单步调试,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-05-05

最新评论