Spring boot 集成 RabbitMQ

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相關依賴

XML

 org.springframework.boot spring-boot-starter-amqp

每錯,就是點配置,不過這樣可能有點不理解,我還是把全部配置貼出來吧

XML

  4.0.0 wang.raye.rabbitmq  < artifactId >demo1 0.0.1-SNAPSHOT  jar  demo1   maven.apache.org  UTF-8  < parent > org.springframework.boot groupId > spring-boot-starter-parent artifactId > 1.4.0.RELEASE version >  < dependencies >  < groupId >junit < artifactId >junit < version >3.8.1 < scope >test  dependency >  < dependency > org.springframework.boot  spring-boot-starter-web artifactId >  dependency >   < groupId >org.springframework.boot groupId > < artifactId >spring-boot-starter-amqp artifactId >  dependency >  dependencies >  project > 

因為沒有做其他操作,所以目前項目主要是依賴2個模塊,一個Sprig boot,一個RabbitMQ

添加配置類

Java

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*/@Configurationpublic 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,通過訪問接口的形式來調用發送消息的方法,話不多說,上代碼

Java

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**/@RestControllerpublic 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


分享到:


相關文章: