Spring boot集成RabbitMQ
作者：Raye；
鏈接：https://www.jianshu.com/p/6ca34345b796
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項目在前面的文章已經說過很多次了，所以這裏就不多說了，如果不會可以去看其他關於spring boot的博客
添加RabbitMQ相關依賴
org.springframework.boot
spring-boot-starter-amqp
每錯，就是點配置，不過這樣可能有點不理解，我還是把全部配置貼出來吧
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
wang.raye.rabbitmq
demo1
0.0.1-SNAPSHOT
jar
demo1
http://maven.apache.org
UTF-8
org.springframework.boot
spring-boot-starter-parent
1.4.0.RELEASE
junit
junit
3.8.1
test
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-amqp
因爲沒有做其他操作，所以目前項目主要是依賴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
https://github.com/RayeWang/rabbitmqdemo
