Spring boot集成RabbitMQ的示例代码

rabbitmq简介

rabbitmq是一个在amqp基础上完整的，可复用的企业消息系统

mq全称为message queue, 消息队列（mq）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

amqp就是一个协议，是一个高级抽象层消息通信协议。

虽然在同步消息通讯的世界里有很多公开标准（如 cobar的 iiop ，或者是 soap 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 msmq ，ibm 的 websphere mq 等），因此，在 2006 年的 6 月，cisco 、redhat、imatix 等联合制定了 amqp 的公开标准。也就是说amqp是异步通讯的一个协议。

rabbitmq使用场景

在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。不过大多数不仅仅是无需即时返回，甚至是执行是否成功都无所谓。如果需要即时返回则可以使用dubbo,spring boot与dubbo集成可以去看spring boot 集成dubbox

rabbitmq依赖

rabbitmq并不是直接一个简单的jar包（jar包只是提供一个基本的与rabbitmq本身通讯的一些功能），和dubbo相同，rabbitmq也需要其他软件来运行，以下是rabbitmq运行所需要的软件

1、erlang

由于rabbitmq软件本身是基于erlang开发的，所以想要运行rabbitmq必须要先按照erlang

erlang官网

erlang下载地址

rabbitmq

rabbitmq才是实现消息队列的核心

rabbitmq官网

rabbitmq下载

配置rabbitmq

安装完成后，需要完成一些配置才能使用rabbitmq，可以直接用cmd到rabbitmq的安装目录下的sbin目录通过命令配置，也可以直接在开始菜单中直接找到rabbitmq command prompt (sbin dir)运行直接到达rabbitmq的安装目录的sbin，为了方便，我们先启用管理插件,执行命令

rabbitmq-plugins.bat enable rabbitmq_management 

即可，注意，这是在windows下面，如果是linux则没有bat后缀 然后我们添加一个用户，因为在外网环境没有用户的情况下是不能连接成功的，执行添加用户命令

rabbitmqctl.bat add_user springboot password 

springboot是用户名，password是密码

然后为了方便演示，我们给springboot赋予管理员权限，方便登录管理页面

rabbitmqctl.bat set_user_tags springboot administrator 

给账号赋予虚拟主机权限

rabbitmqctl.bat set_permissions -p / springboot .* .* .* 

然后启动rabbitmq服务 访问rabbitmq管理页面http://localhost:15672即可看见登录页面，如果没有创建用户则可以用guest，guest登录，如果有创建用户则用创建的用户登录

创建springboot项目

因为创建spring boot项目在前面的文章已经说过很多次了，所以这里就不多说了

添加rabbitmq相关依赖

    <!-- rabbitmq -->
    <dependency>
      <groupid>org.springframework.boot</groupid>
      <artifactid>spring-boot-starter-amqp</artifactid>
    </dependency>

没错，就是点配置，不过这样可能有点不理解，我还是把全部配置贴出来吧

<project xmlns="http://maven.apache.org/pom/4.0.0" xmlns:xsi="http://www.w3.org/2001/xmlschema-instance" 
 xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelversion>4.0.0</modelversion>

 <groupid>wang.raye.rabbitmq</groupid>
 <artifactid>demo1</artifactid>
 <version>0.0.1-snapshot</version>
 <packaging>jar</packaging>

 <name>demo1</name>
 <url>http://maven.apache.org</url>

 <properties>
  <project.build.sourceencoding>utf-8</project.build.sourceencoding>
 </properties>
<parent> 
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-parent</artifactid>
    <version>1.4.0.release</version>
  </parent>

  <dependencies>
    <dependency>
      <groupid>junit</groupid>
      <artifactid>junit</artifactid>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <!-- springboot -->
    <dependency>
      <groupid>org.springframework.boot</groupid>
      <artifactid>spring-boot-starter-web</artifactid>

    </dependency>
    <!-- rabbitmq -->
    <dependency>
      <groupid>org.springframework.boot</groupid>
      <artifactid>spring-boot-starter-amqp</artifactid>
    </dependency>
  </dependencies>
</project>

因为没有做其他操作，所以目前项目主要是依赖2个模块，一个sprig boot，一个rabbitmq

添加配置类

package wang.raye.rabbitmq.demo1;
import org.springframework.amqp.core.acknowledgemode; 
import org.springframework.amqp.core.binding; 
import org.springframework.amqp.core.bindingbuilder; 
import org.springframework.amqp.core.directexchange; 
import org.springframework.amqp.core.message; 
import org.springframework.amqp.core.queue; 
import org.springframework.amqp.rabbit.connection.cachingconnectionfactory; 
import org.springframework.amqp.rabbit.connection.connectionfactory; 
import org.springframework.amqp.rabbit.core.channelawaremessagelistener; 
import org.springframework.amqp.rabbit.listener.simplemessagelistenercontainer; 
import org.springframework.context.annotation.bean; 
import org.springframework.context.annotation.configuration;
/**
 * rabbitmq 的配置类
 * 
 * @author raye
 * @since 2016年10月12日10:57:44
 */
@configuration 
public class rabbitmqconfig { 
  /** 消息交换机的名字*/
  public static final string exchange = "my-mq-exchange";
  /** 队列key1*/
  public static final string routingkey1 = "queue_one_key1";
  /** 队列key2*/
  public static final string routingkey2 = "queue_one_key2";

  /**
   * 配置链接信息
   * @return
   */
  @bean
  public connectionfactory connectionfactory() {
    cachingconnectionfactory connectionfactory = new cachingconnectionfactory("127.0.0.1",5672);

    connectionfactory.setusername("springboot");
    connectionfactory.setpassword("password");
    connectionfactory.setvirtualhost("/");
    connectionfactory.setpublisherconfirms(true); // 必须要设置
    return connectionfactory;
  }

  /** 
   * 配置消息交换机
   * 针对消费者配置 
    fanoutexchange: 将消息分发到所有的绑定队列，无routingkey的概念 
    headersexchange ：通过添加属性key-value匹配 
    directexchange:按照routingkey分发到指定队列 
    topicexchange:多关键字匹配 
   */ 
  @bean 
  public directexchange defaultexchange() { 
    return new directexchange(exchange, true, false);
  } 
  /**
   * 配置消息队列1
   * 针对消费者配置 
   * @return
   */
  @bean 
  public queue queue() { 
    return new queue("queue_one", true); //队列持久 

  }
  /**
   * 将消息队列1与交换机绑定
   * 针对消费者配置 
   * @return
   */
  @bean 
  public binding binding() { 
    return bindingbuilder.bind(queue()).to(defaultexchange()).with(rabbitmqconfig.routingkey1); 
  } 

  /**
   * 配置消息队列2
   * 针对消费者配置 
   * @return
   */
  @bean 
  public queue queue1() { 
    return new queue("queue_one1", true); //队列持久 

  }
  /**
   * 将消息队列2与交换机绑定
   * 针对消费者配置 
   * @return
   */
  @bean 
  public binding binding1() { 
    return bindingbuilder.bind(queue1()).to(defaultexchange()).with(rabbitmqconfig.routingkey2); 
  } 
  /**
   * 接受消息的监听，这个监听会接受消息队列1的消息
   * 针对消费者配置 
   * @return
   */
  @bean 
  public simplemessagelistenercontainer messagecontainer() { 
    simplemessagelistenercontainer container = new simplemessagelistenercontainer(connectionfactory()); 
    container.setqueues(queue()); 
    container.setexposelistenerchannel(true); 
    container.setmaxconcurrentconsumers(1); 
    container.setconcurrentconsumers(1); 
    container.setacknowledgemode(acknowledgemode.manual); //设置确认模式手工确认 
    container.setmessagelistener(new channelawaremessagelistener() {
      public void onmessage(message message, com.rabbitmq.client.channel channel) throws exception {
        byte[] body = message.getbody(); 
        system.out.println("收到消息 : " + new string(body)); 
        channel.basicack(message.getmessageproperties().getdeliverytag(), false); //确认消息成功消费 

      } 

    }); 
    return container; 
  } 
  /**
   * 接受消息的监听，这个监听会接受消息队列1的消息
   * 针对消费者配置 
   * @return
   */
  @bean 
  public simplemessagelistenercontainer messagecontainer2() { 
    simplemessagelistenercontainer container = new simplemessagelistenercontainer(connectionfactory()); 
    container.setqueues(queue1()); 
    container.setexposelistenerchannel(true); 
    container.setmaxconcurrentconsumers(1); 
    container.setconcurrentconsumers(1); 
    container.setacknowledgemode(acknowledgemode.manual); //设置确认模式手工确认 
    container.setmessagelistener(new channelawaremessagelistener() {

      public void onmessage(message message, com.rabbitmq.client.channel channel) throws exception {
        byte[] body = message.getbody(); 
        system.out.println("queue1 收到消息 : " + new string(body)); 
        channel.basicack(message.getmessageproperties().getdeliverytag(), false); //确认消息成功消费 
      } 
    }); 
    return container; 
  } 
}

注意，为了更好的展示如何配置，我配置了2个消息队列，而本类除了链接配置哪里，其他都是针对消息消费者的，当然不管消息消费者和消息生产者都需要配置链接信息，而为了方便，所以本项目的消息消费者和生产者都在本项目，一般实际项目中不会在同一项目，由于注释很详细，我就不多说了

发送消息

为了方便发送消息，所以我直接写了一个controller，通过访问接口的形式来调用发送消息的方法，话不多说，上代码

package wang.raye.rabbitmq.demo1;
import java.util.uuid;
import org.springframework.amqp.rabbit.core.rabbittemplate; 
import org.springframework.amqp.rabbit.support.correlationdata; 
import org.springframework.web.bind.annotation.requestmapping; 
import org.springframework.web.bind.annotation.restcontroller;

/**
 * 测试rabbitmq发送消息的controller
 * @author raye
 *
 */
@restcontroller
public class sendcontroller implements rabbittemplate.confirmcallback{ 
  private rabbittemplate rabbittemplate;
  /**
   * 配置发送消息的rabbittemplate，因为是构造方法，所以不用注解spring也会自动注入（应该是新版本的特性）
   * @param rabbittemplate
   */
  public sendcontroller(rabbittemplate rabbittemplate){
    this.rabbittemplate = rabbittemplate;
    //设置消费回调
    this.rabbittemplate.setconfirmcallback(this);
  }
  /**
   * 向消息队列1中发送消息
   * @param msg
   * @return
   */
  @requestmapping("send1")
  public string send1(string msg){
    string uuid = uuid.randomuuid().tostring();
    correlationdata correlationid = new correlationdata(uuid);
    rabbittemplate.convertandsend(rabbitmqconfig.exchange, rabbitmqconfig.routingkey1, msg,
        correlationid);
    return null;
  }
  /**
   * 向消息队列2中发送消息
   * @param msg
   * @return
   */
  @requestmapping("send2")
  public string send2(string msg){
    string uuid = uuid.randomuuid().tostring();
    correlationdata correlationid = new correlationdata(uuid);
    rabbittemplate.convertandsend(rabbitmqconfig.exchange, rabbitmqconfig.routingkey2, msg,
        correlationid);
    return null;
  }
  /**
   * 消息的回调，主要是实现rabbittemplate.confirmcallback接口
   * 注意，消息回调只能代表成功消息发送到rabbitmq服务器，不能代表消息被成功处理和接受
   */
  public void confirm(correlationdata correlationdata, boolean ack, string cause) {
    system.out.println(" 回调id:" + correlationdata);
    if (ack) {
      system.out.println("消息成功消费");
    } else {
      system.out.println("消息消费失败:" + cause+"\n重新发送");
    }
  }
}

需要注意的是消息回调只能代表消息成功发送到rabbitmq服务器

然后我们启动项目，访问http://localhost:8082/send1?msg=aaaa 会发现控制台输出了

收到消息 : aaaa
 回调id:correlationdata [id=37e6e913-835a-4eca-98d1-807325c5900f]
消息成功消费

当然回调id可能不同，如果我们访问http://localhost:8082/send2?msg=bbbb 则输出

queue1 收到消息 : bbbb 
 回调id:correlationdata [id=0cec7500-3117-4aa2-9ea5-4790879812d4]
消息成功消费

最后说两句

因为本文主要是说明如何从零到springboot集成rabbitmq，所以对于rabbitmq的很多信息和用法没有说明，如果对rabbitmq本身不太熟悉的可以去看看其他关于rabbitmq的文章,附上本文demo

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