Windows下RabbitMQ安裝及入門
1.Windows下安裝RabbitMQ需要以下幾個步驟
(1):下載erlang,原因在於RabbitMQ服務端程式碼是使用併發式語言erlang編寫的,下載地址:http://www.erlang.org/downloads,雙擊.exe檔案進行安裝就好,安裝完成之後建立一個名為ERLANG_HOME的環境變數,其值指向erlang的安裝目錄,同時將%ERLANG_HOME%\bin加入到Path中,最後開啟命令列,輸入erl,如果出現erlang的版本資訊就表示erlang語言環境安裝成功;
(2):下載RabbitMQ,下載地址:http://www.rabbitmq.com/,同樣雙擊.exe進行安裝就好(這裡需要注意一點,預設的安裝目錄是C:/Program Files/....,這個目錄中是存在空格符的,我們需要改變安裝目錄,貌似RabbitMQ安裝目錄中是不允許有空格的,我之前踩過這個大坑);
(3):安裝RabbitMQ-Plugins,我的是3.6.*,發現不需要這一步,安裝RabbitMQ.exe檔案後會自動安裝外掛,3.6+的版本跳過該步,直接驗證是否安裝成功。這個相當於是一個管理介面,方便我們在瀏覽器介面檢視RabbitMQ各個訊息佇列以及exchange的工作情況,安裝方法是:開啟命令列cd進入rabbitmq的sbin目錄(我的目錄是:E:\software\rabbitmq\rabbitmq_server-3.6.5\sbin),輸入:rabbitmq-plugins enable rabbitmq_management命令,稍等會會發現出現plugins安裝成功的提示,預設是安裝6個外掛,如果你在安裝外掛的過程中出現了下面的錯誤:
解決方法是:首先在命令列輸入:rabbitmq-service stop,接著輸入rabbitmq-service remove,再接著輸入rabbitmq-service install,接著輸入rabbitmq-service start,最後重新輸入rabbitmq-plugins enable rabbitmq_management試試,我是這樣解決的;
(4)安裝成功驗證。安裝成功後,可以通過dos命令來檢視是否真的安裝成功且正常啟動,因此cmd後輸入以下命令:-- 定位到rabbitmq的安裝路徑:
-- 上述命令回車後接著輸入rabbitmqctl status,回車後出現下面一坨的即說明安裝沒有問題:
-- 此時,已經可以通過 http://127.0.0.1:15672/ 地址來訪問web管理介面了,預設的賬戶和密碼均是 guest。
(5):外掛安裝完之後,在瀏覽器輸入http://localhost:15672進行驗證,你會看到下面介面,輸入使用者名稱:guest,密碼:guest你就可以進入管理介面,當然使用者名稱密碼你都可以變的;
2.安裝完RabbitMQ之後,我們先來簡單瞭解下RabbitMQ中涉及到的幾個概念
producer:訊息生產者
consumer:訊息消費者
virtual host:虛擬主機,在RabbitMQ中,使用者只能在虛擬主機的層面上進行一些許可權設定,比如我可以訪問哪些佇列,我可以處理哪些請求等等;
broker:訊息轉發者,也就是我們RabbitMQ服務端充當的功能了,那麼訊息是按照什麼規則進行轉發的呢?需要用到下面幾個概念;
exchange:交換機,他是和producer直接進行打交道的,有點類似於路由器的功能,主要就是進行轉發操作的唄,那麼producer到底用哪個exchange進行路由呢?這個取決於routing key(路由鍵),每個訊息都有這個鍵,我們也可以自己設定,其實就是一字串;
queue:訊息佇列,用於存放訊息,他接收exchange路由過來的訊息,我們可以對佇列內容進行持久化操作,那麼queue到底接收那個exchange路由的訊息呢?這個時候就要用到binding key(繫結鍵)了,繫結鍵會將佇列和exchange進行繫結,至於繫結方式,RabbitMQ提供了多種方式,大家可以看看鴻洋大神的RabbitMQ部落格系列(點選檢視);
以上就是RabbitMQ涉及到的一些概念了,用一張圖表示這些概念之間的關係就是:
3.RabbitMQ簡單使用
producer(生產者)端步驟:
(1):建立ConnectionFactory,並且設定一些引數,比如hostname,portNumber等等
(2):利用ConnectionFactory建立一個Connection連線
(3):利用Connection建立一個Channel通道
(4):建立queue並且和Channel進行繫結
(5):建立訊息,並且傳送到佇列中
注意,在我們當前的例子中,並沒有用到exchange交換機,RabbitMQ預設情況下是會建立一個空字串名字的exchange的,如果我們沒有建立自己的exchange的話,預設就是使用的這個exchange;
producer端程式碼:
- publicclass Sender {
- privatefinalstatic String QUEUE_NAME = "MyQueue";
- publicstaticvoid main(String[] args) {
- send();
- }
- publicstaticvoid send()
- {
- ConnectionFactory factory = null;
- Connection connection = null;
- Channel channel = null;
- try {
- factory = new ConnectionFactory();
- factory.setHost("localhost");
- connection = factory.newConnection();
- channel = connection.createChannel();
- channel.queueDeclare(QUEUE_NAME, false, false, false, null);
- String message = "my first message .....";
- channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
- System.out.println("已經發送訊息....."+message);
- } catch (IOException e) {
- e.printStackTrace();
- } catch (TimeoutException e) {
- e.printStackTrace();
- }finally{
- try {
- //關閉資源
- channel.close();
- connection.close();
- } catch (IOException e) {
- e.printStackTrace();
- } catch (TimeoutException e) {
- e.printStackTrace();
- }
- }
- }
- }
consumer(消費者)端步驟:
(1):建立ConnectionFactory,並且設定一些引數,比如hostname,portNumber等等
(2):利用ConnectionFactory建立一個Connection連線
(3):利用Connection建立一個Channel通道
(4):將queue和Channel進行繫結,注意這裡的queue名字要和前面producer建立的queue一致
(5):建立消費者Consumer來接收訊息,同時將消費者和queue進行繫結
consumer端程式碼:
- publicclass Receiver {
- privatefinalstatic String QUEUE_NAME = "MyQueue";
- publicstaticvoid main(String[] args) {
- receive();
- }
- publicstaticvoid receive()
- {
- ConnectionFactory factory = null;
- Connection connection = null;
- Channel channel = null;
- try {
- factory = new ConnectionFactory();
- factory.setHost("localhost");
- connection = factory.newConnection();
- channel = connection.createChannel();
- channel.queueDeclare(QUEUE_NAME, false, false, false, null);
- Consumer consumer = new DefaultConsumer(channel){
- @Override
- publicvoid handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
- byte[] body) throws IOException {
- System.out.println("11111111111");
- String message = new String(body, "UTF-8");
- System.out.println("收到訊息....."+message);
- }};
- channel.basicConsume(QUEUE_NAME, true,consumer);
- } catch (IOException e) {
- e.printStackTrace();
- } catch (TimeoutException e) {
- e.printStackTrace();
- }finally{
- try {
- //關閉資源
- channel.close();
- connection.close();
- } catch (IOException e) {
- e.printStackTrace();
- } catch (TimeoutException e) {
- e.printStackTrace();
- }
- }
- }
- }