詳解spring boot集成RabbitMQ

更新時間：2017年03月30日 10:11:15 作者：SamHxm

RabbitMQ作為AMQP的代表性產(chǎn)品，在項目中大量使用。結(jié)合現(xiàn)在主流的spring boot，極大簡化了開發(fā)過程中所涉及到的消息通信問題。

首先正確的安裝RabbitMQ及運行正常。

RabbitMQ需啊erlang環(huán)境，所以首先安裝對應(yīng)版本的erlang，可在RabbitMQ官網(wǎng)下載

# 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默認(rèn)提供的guest用戶只能本地訪問，所以額外創(chuàng)建用戶用于測試

# /sbin/rabbitmqctl add_user test test123
用戶名：test，密碼：test123

開啟web管理插件

# rabbitmq-plugins enable rabbitmq_management

并使用之前創(chuàng)建的用戶登錄,并設(shè)置該用戶為administrator,虛擬主機地址為/

spring boot 引入相關(guān)依賴

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

消息生產(chǎn)者

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隊列分別綁定。

生產(chǎn)者類

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("消息發(fā)送成功:" + correlationData); 
    } else { 
      System.out.println("消息發(fā)送失敗:" + cause); 
    } 

  }

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

  }

  //發(fā)送消息，不需要實現(xiàn)任何接口，供外部調(diào)用。
  public void send(String msg){

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

    System.out.println("開始發(fā)送消息 : " + msg.toLowerCase());
    String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
    System.out.println("結(jié)束發(fā)送消息 : " + msg.toLowerCase());
    System.out.println("消費者響應(yīng) : " + response + " 消息處理完成");
  }
}

要點：

1.注入RabbitTemplate

2.實現(xiàn)RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口（非必須）。
ConfirmCallback接口用于實現(xiàn)消息發(fā)送到RabbitMQ交換器后接收ack回調(diào)。ReturnCallback接口用于實現(xiàn)消息發(fā)送到RabbitMQ交換器，但無相應(yīng)隊列與交換器綁定時的回調(diào)。

3.實現(xiàn)消息發(fā)送方法。調(diào)用rabbitTemplate相應(yīng)的方法即可。rabbitTemplate常用發(fā)送方法有

rabbitTemplate.send(message);  //發(fā)消息，參數(shù)類型為org.springframework.amqp.core.Message
rabbitTemplate.convertAndSend(object); //轉(zhuǎn)換并發(fā)送消息。 將參數(shù)對象轉(zhuǎn)換為org.springframework.amqp.core.Message后發(fā)送
rabbitTemplate.convertSendAndReceive(message) //轉(zhuǎn)換并發(fā)送消息,且等待消息者返回響應(yīng)消息。

針對業(yè)務(wù)場景選擇合適的消息發(fā)送方式即可。

消息消費者

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  //最小消息監(jiān)聽線程數(shù)
spring.rabbitmq.listener.max-concurrency=2 //最大消息監(jiān)聽線程數(shù)

消費者類

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個隊列，所以分別定義不同的監(jiān)聽器監(jiān)聽不同的隊列。由于最小消息監(jiān)聽線程數(shù)和最大消息監(jiān)聽線程數(shù)都是2，所以每個監(jiān)聽器各有2個線程實現(xiàn)監(jiān)聽功能。

要點:

1.監(jiān)聽器參數(shù)類型與消息實際類型匹配。在生產(chǎn)者中發(fā)送的消息實際類型是String，所以這里監(jiān)聽器參數(shù)類型也是String。

2.如果監(jiān)聽器需要有響應(yīng)返回給生產(chǎn)者，直接在監(jiān)聽方法中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);
    }
  }
}

輸出:

開始發(fā)送消息 : 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
結(jié)束發(fā)送消息 : wed mar 29 23:20:52 cst 2017
消費者響應(yīng) : WED MAR 29 23:20:52 CST 2017 消息處理完成
------------------------------------------------
消息發(fā)送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886]
開始發(fā)送消息 : 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
結(jié)束發(fā)送消息 : wed mar 29 23:20:53 cst 2017
消費者響應(yīng) : WED MAR 29 23:20:53 CST 2017 消息處理完成
------------------------------------------------
消息發(fā)送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34]
開始發(fā)送消息 : 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
結(jié)束發(fā)送消息 : wed mar 29 23:20:54 cst 2017
消費者響應(yīng) : WED MAR 29 23:20:54 CST 2017 消息處理完成
------------------------------------------------

如果需要使用的其他的交換器類型，spring中都已提供實現(xiàn)，所有的交換器均實現(xiàn)org.springframework.amqp.core.AbstractExchange接口。

常用交換器類型如下：

Direct(DirectExchange)：direct 類型的行為是"先匹配, 再投送". 即在綁定時設(shè)定一個 routing_key, 消息的routing_key完全匹配時, 才會被交換器投送到綁定的隊列中去。

Topic(TopicExchange)：按規(guī)則轉(zhuǎn)發(fā)消息（最靈活）。

Headers(HeadersExchange)：設(shè)置header attribute參數(shù)類型的交換機。

Fanout(FanoutExchange)：轉(zhuǎn)發(fā)消息到所有綁定隊列。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: