Spring boot集成RabbitMQ的示例代碼

rabbitmq-plugins.bat enable rabbitmq_management 

rabbitmqctl.bat add_user springboot password 

rabbitmqctl.bat set_user_tags springboot administrator 

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

    <!-- 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>

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 的配置類(lèi)
 * 
 * @author Raye
 * @since 2016年10月12日10:57:44
 */
@Configuration 
public class RabbitMQConfig { 
  /** 消息交換機(jī)的名字*/
  public static final String EXCHANGE = "my-mq-exchange";
  /** 隊(duì)列key1*/
  public static final String ROUTINGKEY1 = "queue_one_key1";
  /** 隊(duì)列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); // 必須要設(shè)置
    return connectionFactory;
  }

  /** 
   * 配置消息交換機(jī)
   * 針對(duì)消費(fèi)者配置 
    FanoutExchange: 將消息分發(fā)到所有的綁定隊(duì)列，無(wú)routingkey的概念 
    HeadersExchange ：通過(guò)添加屬性key-value匹配 
    DirectExchange:按照routingkey分發(fā)到指定隊(duì)列 
    TopicExchange:多關(guān)鍵字匹配 
   */ 
  @Bean 
  public DirectExchange defaultExchange() { 
    return new DirectExchange(EXCHANGE, true, false);
  } 
  /**
   * 配置消息隊(duì)列1
   * 針對(duì)消費(fèi)者配置 
   * @return
   */
  @Bean 
  public Queue queue() { 
    return new Queue("queue_one", true); //隊(duì)列持久 

  }
  /**
   * 將消息隊(duì)列1與交換機(jī)綁定
   * 針對(duì)消費(fèi)者配置 
   * @return
   */
  @Bean 
  public Binding binding() { 
    return BindingBuilder.bind(queue()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY1); 
  } 

  /**
   * 配置消息隊(duì)列2
   * 針對(duì)消費(fèi)者配置 
   * @return
   */
  @Bean 
  public Queue queue1() { 
    return new Queue("queue_one1", true); //隊(duì)列持久 

  }
  /**
   * 將消息隊(duì)列2與交換機(jī)綁定
   * 針對(duì)消費(fèi)者配置 
   * @return
   */
  @Bean 
  public Binding binding1() { 
    return BindingBuilder.bind(queue1()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY2); 
  } 
  /**
   * 接受消息的監(jiān)聽(tīng)，這個(gè)監(jiān)聽(tīng)會(huì)接受消息隊(duì)列1的消息
   * 針對(duì)消費(fèi)者配置 
   * @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); //設(shè)置確認(rèn)模式手工確認(rèn) 
    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); //確認(rèn)消息成功消費(fèi) 

      } 

    }); 
    return container; 
  } 
  /**
   * 接受消息的監(jiān)聽(tīng)，這個(gè)監(jiān)聽(tīng)會(huì)接受消息隊(duì)列1的消息
   * 針對(duì)消費(fèi)者配置 
   * @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); //設(shè)置確認(rèn)模式手工確認(rèn) 
    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); //確認(rèn)消息成功消費(fèi) 
      } 
    }); 
    return container; 
  } 
}

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;

/**
 * 測(cè)試RabbitMQ發(fā)送消息的Controller
 * @author Raye
 *
 */
@RestController
public class SendController implements RabbitTemplate.ConfirmCallback{ 
  private RabbitTemplate rabbitTemplate;
  /**
   * 配置發(fā)送消息的rabbitTemplate，因?yàn)槭菢?gòu)造方法，所以不用注解Spring也會(huì)自動(dòng)注入（應(yīng)該是新版本的特性）
   * @param rabbitTemplate
   */
  public SendController(RabbitTemplate rabbitTemplate){
    this.rabbitTemplate = rabbitTemplate;
    //設(shè)置消費(fèi)回調(diào)
    this.rabbitTemplate.setConfirmCallback(this);
  }
  /**
   * 向消息隊(duì)列1中發(fā)送消息
   * @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;
  }
  /**
   * 向消息隊(duì)列2中發(fā)送消息
   * @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;
  }
  /**
   * 消息的回調(diào)，主要是實(shí)現(xiàn)RabbitTemplate.ConfirmCallback接口
   * 注意，消息回調(diào)只能代表成功消息發(fā)送到RabbitMQ服務(wù)器，不能代表消息被成功處理和接受
   */
  public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    System.out.println(" 回調(diào)id:" + correlationData);
    if (ack) {
      System.out.println("消息成功消費(fèi)");
    } else {
      System.out.println("消息消費(fèi)失敗:" + cause+"\n重新發(fā)送");
    }
  }
}