快捷導(dǎo)航

Spring Boot系列教程之7步集成RabbitMQ的方法

更新時(shí)間：2018年11月12日 11:44:59 作者：JackieZheng

RabbitMQ 即一個(gè)消息隊(duì)列，主要是用來(lái)實(shí)現(xiàn)應(yīng)用程序的異步和解耦，同時(shí)也能起到消息緩沖，消息分發(fā)的作用。下面這篇文章主要給大家介紹了關(guān)于Spring Boot之7步集成RabbitMQ的相關(guān)資料，需要的朋友可以參考下

前言

RabbitMQ是一種我們經(jīng)常使用的消息中間件，RabbitMQ是實(shí)現(xiàn)AMQP（高級(jí)消息隊(duì)列協(xié)議）的消息中間件的一種，最初起源于金融系統(tǒng)，用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。RabbitMQ主要是為了實(shí)現(xiàn)系統(tǒng)之間的雙向解耦而實(shí)現(xiàn)的。當(dāng)生產(chǎn)者大量產(chǎn)生數(shù)據(jù)時(shí)，消費(fèi)者無(wú)法快速消費(fèi)，那么需要一個(gè)中間層。保存這個(gè)數(shù)據(jù)。

AMQP，即Advanced Message Queuing Protocol，高級(jí)消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開(kāi)放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無(wú)需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對(duì)點(diǎn)和發(fā)布/訂閱）、可靠性、安全。

RabbitMQ是一個(gè)開(kāi)源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語(yǔ)言編寫(xiě)，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。

今天這篇，我們來(lái)看看Spring Boot是如何集成RabbitMQ，發(fā)送消息和消費(fèi)消息的。同時(shí)我們介紹下死信隊(duì)列。

集成RabbitMQ

集成RabbitMQ只需要如下幾步即可

1、添加maven依賴(lài)

<!--rabbitmq-->

<dependency>

 <groupId>org.springframework.boot</groupId>

 <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

2、添加配置文件application.yaml

在application.yaml添加配置內(nèi)容如下

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四個(gè)屬性即可

其他屬性都有各自的含義，比如retry是用于配置重試策略的，acknowledge-mode是配置消息接收確認(rèn)機(jī)制的。

3、編寫(xiě)配置類(lèi)

編寫(xiě)RabbitConfig配置類(lèi)，采用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;
 } // --------------------- 聲明隊(duì)列 ------------------------
 @Bean
 public Queue demoQueue() {
 return new Queue("demo_queue");
 } // --------------------- 聲明exchange ------------------------ @Bean
 public DirectExchange demoExchange() {
 return new DirectExchange("demo_exchange");
 } // --------------------- 隊(duì)列綁定 ------------------------
 @Bean
 public Binding bindingAlbumItemCreatedQueue(DirectExchange demoExchange,
  Queue demoQueue) {
 return BindingBuilder.bind(demoQueue).to(demoExchange).with("100");
 } }

注意

這里聲明了Direct模式的Exchange，聲明一個(gè)Queue，并通過(guò)routing-key為100將demo_queue綁定到demo_exchange，這樣demo_queue就可以接收到demo_exchange發(fā)送的消息了。

4、編寫(xiě)消息發(fā)送類(lèi)

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;

 /**
 * 構(gòu)造方法注入 */ @Autowired
 public Sender(RabbitTemplate rabbitTemplate) {
 this.rabbitTemplate = rabbitTemplate;
 rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果為單例的話，那回調(diào)就是最后設(shè)置的內(nèi)容
 } public void sendMsg(String content) {
 rabbitTemplate.convertAndSend("demo_exchange", "100", content);
 } /**
 * 回調(diào) */ @Override
 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);
 }
 } }

注意

發(fā)送內(nèi)容content，路由到routing-key為100上，則我們就可以在demo_queue隊(duì)列中看到發(fā)送的消息內(nèi)容了

confirm函數(shù)是回調(diào)函數(shù)，這里因?yàn)闆](méi)有消費(fèi)者，且acknoledge-mode是auto(其他兩種值分別是none和manual)，所以ack是false。

5、編寫(xiě)發(fā)送消息測(cè)試類(lèi)

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");
 } }

運(yùn)行該測(cè)試類(lèi)，我們可以看到如下結(jié)果

6、編寫(xiě)消息消費(fèi)類(lèi)

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);
 } }

注意

消息消費(fèi)類(lèi)也非常簡(jiǎn)單，添加注解@RabbitListener，指定要監(jiān)聽(tīng)的隊(duì)列名稱(chēng)即可

除了注解@RabbitListener，我們經(jīng)常還能看到@RabbitHandler，這兩個(gè)注解可以配合起來(lái)使用。

@RabbitListener 標(biāo)注在類(lèi)上面表示當(dāng)有收到消息的時(shí)候，就交給 @RabbitHandler 的方法處理，具體使用哪個(gè)方法處理，根據(jù) MessageConverter 轉(zhuǎn)換后的參數(shù)類(lèi)型，形如

@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、運(yùn)行消息發(fā)送測(cè)試類(lèi)