關於RabbitMQ以及RabbitMQ和Spring的整合
基本概念
RabbitMQ是流行的開源訊息佇列系統,用erlang語言開發。RabbitMQ是AMQP(高階訊息佇列協議)的標準實現。如果不熟悉AMQP,直接看RabbitMQ的文件會比較困難。不過它也只有幾個關鍵概念,這裡簡單介紹。
RabbitMQ的結構圖如下:
幾個概念說明:
Broker:簡單來說就是訊息佇列伺服器實體。 Exchange:訊息交換機,它指定訊息按什麼規則,路由到哪個佇列。 Queue:訊息佇列載體,每個訊息都會被投入到一個或多個佇列。 Binding:繫結,它的作用就是把exchange和queue按照路由規則繫結起來。 Routing Key:路由關鍵字,exchange根據這個關鍵字進行訊息投遞。 vhost:虛擬主機,一個broker裡可以開設多個vhost,用作不同使用者的許可權分離。 producer:訊息生產者,就是投遞訊息的程式。 consumer:訊息消費者,就是接受訊息的程式。 channel:訊息通道,在客戶端的每個連線裡,可建立多個channel,每個channel代表一個會話任務。
訊息佇列的使用過程大概如下:
(1)客戶端連線到訊息佇列伺服器,開啟一個channel。 (2)客戶端宣告一個exchange,並設定相關屬性。 (3)客戶端宣告一個queue,並設定相關屬性。 (4)客戶端使用routing key,在exchange和queue之間建立好繫結關係。 (5)客戶端投遞訊息到exchange。
exchange接收到訊息後,就根據訊息的key和已經設定的binding,進行訊息路由,將訊息投遞到一個或多個佇列裡。
exchange也有幾個型別,完全根據key進行投遞的叫做Direct交換機,例如,繫結時設定了routing key為”abc”,那麼客戶端提交的訊息,只有設定了key為”abc”的才會投遞到佇列。對key進行模式匹配後進行投遞的叫做Topic交換機,符號”#”匹配一個或多個詞,符號”*”匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”,”abc.*”只匹配”abc.def”。還有一種不需要key的,叫做Fanout交換機,它採取廣播模式,一個訊息進來時,投遞到與該交換機繫結的所有佇列。
RabbitMQ支援訊息的持久化,也就是資料寫在磁碟上,為了資料安全考慮,我想大多數使用者都會選擇持久化。訊息佇列持久化包括3個部分: (1)exchange持久化,在宣告時指定durable => 1 (2)queue持久化,在宣告時指定durable => 1 (3)訊息持久化,在投遞時指定delivery_mode => 2(1是非持久化)
如果exchange和queue都是持久化的,那麼它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化,一個非持久化,就不允許建立繫結。
rabbitmq整合spring
1.需要下載rabbitmq, 而在rabbitmq安裝之前需要安裝erlang, 因為rabbitmq是用erlang寫的.(下載好之後配置環境變數)
1.1 啟動rabbitmq伺服器
1.2 在rabbitmq伺服器上建立虛擬主機
1.3 建立使用者
1.4 給使用者新增管理員角色
1.5 設定使用者在虛擬主機上的許可權
set_permissions -p vhostSJJ shijunjie ".*" ".*" ".*"
1.6 安裝基於web的管理外掛
按照上面配置好後,可以在管理介面上看到入下資訊
附:rabbitmq的一些常用命令
rabbitmq的安裝、啟動和停止
rabbitmq-service.bat install rabbitmq-service.bat start rabbitmq-service.bat stop
列出所有queue
rabbitmqctl list_queues
列出指定queue的資訊
rabbitmqctl list_queues [the queue name] messages_ready messages_unacknowledged
列出所有exchange
rabbitmqctl list_exchanges
列出所有binding
rabbitmqctl list_bindings
安裝基於web的管理外掛
rabbitmq-plugins.bat enable rabbitmq_management
關閉節點 # rabbitmqctl stop 2.停止RabbitMQ應用 # rabbitmqctl stop_app 3.啟動RabbitMQ應用 # rabbitmqctl start_app 4.顯示RabbitMQ中介軟體各種資訊 # rabbitmqctl status 5.重置RabbitMQ節點 # rabbitmqctl reset # rabbitmqctl force_reset 從它屬於的任何叢集中移除,從管理資料庫中移除所有資料,例如配置過的使用者和虛擬宿主, 刪除所有持久化的訊息。 force_reset命令和reset的區別是無條件重置節點,不管當前管理資料庫狀態以及叢集的配置。如果資料庫或者叢集配置發生錯誤才使用這個最後 的手段。 注意:只有在停止RabbitMQ應用後,reset和force_reset才能成功。 6.迴圈日誌檔案 # rabbitmqctl rotate_logs[suffix] 7.叢集管理 # rabbitmqctl cluster clusternode…
使用者管理 1.新增使用者 # rabbitmqctl add_user username password 2.刪除使用者 # rabbitmqctl delete_user username 3.修改密碼 # rabbitmqctl change_password username newpassword 4.列出所有使用者 # rabbitmqctl list_users
許可權控制1.建立虛擬主機 # rabbitmqctl add_vhost vhostpath 2.刪除虛擬主機 # rabbitmqctl delete_vhost vhostpath 3.列出所有虛擬主機 # rabbitmqctl list_vhosts 4.設定使用者許可權 # rabbitmqctl set_permissions [-p vhostpath] username regexp regexp regexp 5.清除使用者許可權 # rabbitmqctl clear_permissions [-p vhostpath] username 6.列出虛擬主機上的所有許可權 # rabbitmqctl list_permissions [-p vhostpath] 7.列出使用者許可權
# rabbitmqctl list_user_permissions username
2.建立一個maven專案.然後我們開始配置專案.
2.1 由於是spring整合,我們需要加入spring的依賴.
<dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context-support</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-aop</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-aspects</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-tx</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jdbc</artifactId> <version>3.2.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> <version>3.2.8.RELEASE</version> </dependency>
2.2加入rabbitmq和spring的整合依賴
<!--rabbitmq依賴 --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.3.5.RELEASE</version> </dependency>
2.3定義訊息生產者和訊息傳送者
exchange有幾種,這裡我只測試了兩種, 通過分別定義兩個exchange去繫結direct和topic..
首先, 定義訊息生產者, 通過配置將template連結connect-factory並注入到程式碼中使用.
package me.shijunjie.producer; import java.io.IOException; import javax.annotation.Resource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.stereotype.Service; @Service public class MessageProducer { private Logger logger = LoggerFactory.getLogger(MessageProducer.class); @Resource(name="amqpTemplate") private AmqpTemplate amqpTemplate; @Resource(name="amqpTemplate2") private AmqpTemplate amqpTemplate2; public void sendMessage(Object message) throws IOException { logger.info("to send message:{}", message); amqpTemplate.convertAndSend("queueTestKey", message); amqpTemplate.convertAndSend("queueTestChris", message); amqpTemplate2.convertAndSend("shijj.xxxx.wsdwd", message); } }
然後我們定義訊息消費者, 這裡,我定義了三個消費者, 通過監聽訊息佇列, 分別接受各自所匹配的訊息.
第一個消費者, 接受direct的訊息, 他的exchange為exchangeTest, rout-key為queueTestKey
package me.shijunjie.consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; public class MessageConsumer implements MessageListener { private Logger logger = LoggerFactory.getLogger(MessageConsumer.class); @Override public void onMessage(Message message) { logger.info("consumer receive message------->:{}", message); } }
第二個消費者, 接受direct的訊息(為了測試一個exchange可以傳送多個訊息), 他的exchange為exchangeTest, rout-key為queueTestChris.
package me.shijunjie.consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; public class ChrisConsumer implements MessageListener { private Logger logger = LoggerFactory.getLogger(ChrisConsumer.class); @Override public void onMessage(Message message) { logger.info("chris receive message------->:{}", message); } }
第三個消費者, 接受topic的訊息他的exchange為exchangeTest2, pattern為shijj.*.. .*可以匹配一個, .#可以匹配一個或多個..
package me.shijunjie.consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; public class ShijjConsumer implements MessageListener { private Logger logger = LoggerFactory.getLogger(ShijjConsumer.class); @Override public void onMessage(Message message) { logger.info("shijj receive message------->:{}", message); } }
然後就是關鍵rabbit整合spring的配置檔案.
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd"> <!--配置connection-factory,指定連線rabbit server引數 --> <rabbit:connection-factory id="connectionFactory" virtual-host="vhostSJJ" username="shijunjie" password="wssjj123" host="123.206.228.200" port="5672" /> <!-- <rabbit:connection-factory id="connectionFactory" username="test2" password="test2" host="123.206.228.200" port="5672" /> --> <!--通過指定下面的admin資訊,當前producer中的exchange和queue會在rabbitmq伺服器上自動生成 --> <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" /> <!--定義queue --> <rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /> <!-- 定義direct exchange,繫結queueTest --> <rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false" declared-by="connectAdmin"> <rabbit:bindings> <rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding> </rabbit:bindings> </rabbit:direct-exchange> <!--定義rabbit template用於資料的接收和傳送 --> <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="exchangeTest" /> <!-- 訊息接收者 --> <bean id="messageReceiver" class="me.shijunjie.consumer.MessageConsumer"></bean> <!-- queue litener 觀察 監聽模式 當有訊息到達時會通知監聽在對應的佇列上的監聽物件 --> <rabbit:listener-container connection-factory="connectionFactory"> <rabbit:listener queues="queueTest" ref="messageReceiver" /> </rabbit:listener-container> <!--定義queue --> <rabbit:queue name="queueChris" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /> <!-- 定義direct exchange,繫結queueTest --> <rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false" declared-by="connectAdmin"> <rabbit:bindings> <rabbit:binding queue="queueChris" key="queueTestChris"></rabbit:binding> </rabbit:bindings> </rabbit:direct-exchange> <!-- 訊息接收者 --> <bean id="receiverChris" class="me.shijunjie.consumer.ChrisConsumer"></bean> <!-- queue litener 觀察 監聽模式 當有訊息到達時會通知監聽在對應的佇列上的監聽物件 --> <rabbit:listener-container connection-factory="connectionFactory"> <rabbit:listener queues="queueChris" ref="receiverChris" /> </rabbit:listener-container> <!-- 分隔線 --> <!--配置connection-factory,指定連線rabbit server引數 --> <rabbit:connection-factory id="connectionFactory2" virtual-host="vhostSJJ" username="shijunjie" password="wssjj123" host="123.206.228.200" port="5672" /> <!--通過指定下面的admin資訊,當前producer中的exchange和queue會在rabbitmq伺服器上自動生成 --> <rabbit:admin id="connectAdmin2" connection-factory="connectionFactory2" /> <!--定義queue --> <rabbit:queue name="queueShijj" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin2" /> <!-- 定義direct exchange,繫結queueTest --> <rabbit:topic-exchange name="exchangeTest2" durable="true" auto-delete="false" declared-by="connectAdmin2"> <rabbit:bindings> <rabbit:binding queue="queueShijj" pattern="shijj.#"></rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange> <!--定義rabbit template用於資料的接收和傳送 --> <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory2" exchange="exchangeTest2" /> <!-- 訊息接收者 --> <bean id="recieverShijj" class="me.shijunjie.consumer.ShijjConsumer"></bean> <!-- queue litener 觀察 監聽模式 當有訊息到達時會通知監聽在對應的佇列上的監聽物件 --> <rabbit:listener-container connection-factory="connectionFactory2"> <rabbit:listener queues="queueShijj" ref="recieverShijj" /> </rabbit:listener-container> </beans>
以及我的application.xml
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd"> <import resource="classpath*:rabbitMQ.xml" /> <!-- 掃描指定package下所有帶有如@controller,@services,@resource,@ods並把所註釋的註冊為Spring Beans --> <context:component-scan base-package="me.shijunjie.consumer, me.shijunjie.producer" /> <!-- 啟用annotation功能 --> <context:annotation-config /> <!-- 啟用annotation功能 --> <context:spring-configured /> </beans>
3.全部配置完成之後進行測試
1.編寫測試程式碼
package me.shijunjie.test; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import me.shijunjie.producer.MessageProducer; public class TestClass { private Logger logger = LoggerFactory.getLogger(TestClass.class); private ApplicationContext context = null; @Before public void setUp() throws Exception { context = new ClassPathXmlApplicationContext("application.xml"); } @Test public void should_send_a_amq_message() throws Exception { MessageProducer messageProducer = (MessageProducer) context.getBean("messageProducer"); int a = 100; while (a > 0) { messageProducer.sendMessage("Hello, I am amq sender num :" + a--); try { //暫停一下,好讓訊息消費者去取訊息打印出來 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
執行後