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,username和password四個屬性即可

其他屬性都有各自的含義，比如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

本來准備再說說死信隊列的，限於篇幅，后面再寫吧。

如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕，您的“推薦”將是我最大的寫作動力！如果您想持續關注我的文章，請掃描二維碼，關注JackieZheng的微信公眾號，我會將我的文章推送給您，並和您一起分享我日常閱讀過的優質文章。