基本概念
RabbitMQ是流行的開源消息隊列系統，用erlang語言開發。RabbitMQ是AMQP（高級消息隊列協議）的標准實現。如果不熟悉AMQP，直接看RabbitMQ的文檔會比較困難。不過它也只有幾個關鍵概念，這里簡單介紹。
RabbitMQ的結構圖如下：
幾個概念說明：
Broker：簡單來說就是消息隊列服務器實體。
Exchange：消息交換機，它指定消息按什么規則，路由到哪個隊列。
Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。
Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。
Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。
vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。
producer：消息生產者，就是投遞消息的程序。
consumer：消息消費者，就是接受消息的程序。
channel：消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務。
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。
（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是非持久化）
（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(); } } } }
運行后
