RabbitMQ 安裝使用

安裝

下載 Erlang

RabbitMQ 是用這個語言寫的，這個語言性能好，下載就好了。

下載地址

如果是 Windows 直接點擊 Installer 下載傻瓜式下載。

下載的時候一般不放到默認路徑，自己選一個自己特定的安裝路徑。

下載 RabbitMQ 的服務

下載地址

也是直接點擊看下圖

安裝好后看是否有 RabbitMQ 的服務

如果有這樣的服務那就是成功啦。

ctrl + R 輸入 services.msc

查看服務列表

找到 RabbitMQ 服務，如果在運行，那么就是安裝成功了。

開啟管理 UI

在目錄，打開 sbin 目錄

并且在此目錄下打開命令行，運行 rabbitmq-plugins enable rabbitmq_management 如果沒有什么報錯的話就說明管理 UI 的插件已經(jīng)成功配置。

這時候你只需要打開 localhost:15672/ 即可進入管理頁面，如下：

一開始進入，如果需要密碼的話，那么用戶明和密碼都是 guest

如果你在服務端需要遠程管理，這個賬號是登錄不了的，需要你自己再去創(chuàng)建一個賬號。

RabbitMQ 端口使用一覽圖

比如 15672 是管理 UI 的端口

RabbitMQ 服務就是運行在端口 5672

使用

這里使用的是 Java 版本

消息隊列中的有幾個重要的 專有名詞，一個是生產(chǎn)者（producer）=》 生產(chǎn)消息，一個是消費者（consumer）=》 消費消息，一個是 broker 對消息進行轉(zhuǎn)發(fā)的中間人。還有一個就是 router 路由，如何進行轉(zhuǎn)發(fā)。

輸出最簡單的 Hello World！

輸出 Hello World 就是這樣的一個簡單的模型。

有一個生產(chǎn)者創(chuàng)造一條消息轉(zhuǎn)發(fā)到 Broker （這個是由 RabbitMQ 自動完成），一個消費者消費這條消息。

生產(chǎn)者定義

代碼中有詳細的注釋。

package com.xwhking.testcode.TestRabbitMQ;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;public class Send {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory(); // 創(chuàng)建一個RabbitMQ 的連接工廠factory.setHost("localhost"); // 設置RabbitMQ 的主機地址try (Connection connection = factory.newConnection(); // 創(chuàng)建一個連接Channel channel = connection.createChannel()) { // 創(chuàng)一個管道，用于通信，傳輸消息。channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 生命一個隊列， 第一個參數(shù)是隊列的名字，第二個參數(shù)是是否持久化，第三個參數(shù)是是否獨占隊列，第四個參數(shù)是是否自動刪除，第五個參數(shù)是隊列的屬性。String message = "Hello World!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); // 發(fā)布消息，第一個參數(shù)是交換機名，第二個參數(shù)是隊列名，第三個參數(shù)是消息的屬性，第四個參數(shù)是消息的內(nèi)容。傳輸過程都是以二進制進行傳輸。System.out.println(" [x] Sent '" + message + "'");}}
}

運行以后就可以在管理界面看到了。

消費者消費消息

package com.xwhking.testcode.TestRabbitMQ;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;public class Recv {private final static String QUEUE_NAME = "hello"; // 隊列名字public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory(); // 創(chuàng)建工廠factory.setHost("localhost"); // 設置主機Connection connection = factory.newConnection(); // 獲取一個連接Channel channel = connection.createChannel(); // 獲取通道channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 聲明一個隊列，參數(shù)：隊列名，是否持久化，是否獨占，是否自動刪除，參數(shù)System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), StandardCharsets.UTF_8);System.out.println(" [x] Received '" + message + "'");};// 消費者收到消息的回調(diào)函數(shù)，參數(shù)：消費者標簽，消息，回調(diào)函數(shù)內(nèi)，就是處理消息的代碼channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); // 消費者，參數(shù)：隊列名，是否自動應答，回調(diào)函數(shù)，消費者標簽}
}

消費后管理頁面

擁有的一條消息就被消費啦。

小拓展

消費者不是啟動以后，就是一個阻塞線程嗎，如果沒有消息的來到，那么線程會一直阻塞，那么我是否能夠改一下生產(chǎn)者，能夠隨時輸入消息，讓消費者收到呢。

嗯嗯嗯！

說干就干

修改生產(chǎn)者

package com.xwhking.testcode.TestRabbitMQ;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;
import java.util.Scanner;public class Send {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory(); // 創(chuàng)建一個RabbitMQ 的連接工廠factory.setHost("localhost"); // 設置RabbitMQ 的主機地址try (Connection connection = factory.newConnection(); // 創(chuàng)建一個連接Channel channel = connection.createChannel()) { // 創(chuàng)一個管道，用于通信，傳輸消息。channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 生命一個隊列， 第一個參數(shù)是隊列的名字，第二個參數(shù)是是否持久化，第三個參數(shù)是是否獨占隊列，第四個參數(shù)是是否自動刪除，第五個參數(shù)是隊列的屬性。String message = "Hello World!";Scanner scanner = new Scanner(System.in);while(true){message = scanner.nextLine();channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); // 發(fā)布消息，第一個參數(shù)是交換機名，第二個參數(shù)是隊列名，第三個參數(shù)是消息的屬性，第四個參數(shù)是消息的內(nèi)容。傳輸過程都是以二進制進行傳輸。System.out.println(" [x] Sent '" + message + "'");}}}
}

大家好，我是xwhking，一名技術(shù)愛好者，目前正在全力學習 Java，前端也會一點，如果你有任何疑問請你評論，或者可以加我QQ（2837468248）說明來意！希望能夠與你共同進步