簡單的說就是用來傳輸消息的中間載體，就是將你的信息發(fā)送到接受方，它并不關(guān)心發(fā)送的數(shù)據(jù)是什么。RabbitMQ就是一個(gè)消息中間件。
大家好，我是指北君。
在工作中經(jīng)常會(huì)用到消息隊(duì)列處理各種問題，今天指北君帶領(lǐng)大家來學(xué)一個(gè)很常用到的技術(shù)-RabbitMQ；接下來還會(huì)有關(guān)于RabbitMQ的系列教程。
RabbitMQ是什么
隊(duì)列：一種數(shù)據(jù)結(jié)構(gòu)，先進(jìn)先出。
消息隊(duì)列：簡單的說就是用來進(jìn)行消息傳輸?shù)年?duì)列。
消息中間件：簡單的說就是用來傳輸消息的中間載體，就是將你的信息發(fā)送到接受方，它并不關(guān)心發(fā)送的數(shù)據(jù)是什么。RabbitMQ就是一個(gè)消息中間件。
RabbitMQ的特點(diǎn)
- 可靠性。支持持久化，傳輸確認(rèn)，發(fā)布確認(rèn)等保證了MQ的可靠性。
- 靈活的分發(fā)消息策略。這應(yīng)該是RabbitMQ的一大特點(diǎn)。在消息進(jìn)入MQ前由Exchange(交換機(jī))進(jìn)行路由消息。分發(fā)消息策略有：簡單模式、工作隊(duì)列模式、發(fā)布訂閱模式、路由模式、通配符模式。
- 支持集群。多臺(tái)RabbitMQ服務(wù)器可以組成一個(gè)集群，形成一個(gè)邏輯Broker。
- 多種協(xié)議。RabbitMQ支持多種消息隊(duì)列協(xié)議，比如 STOMP、MQTT 等等。
- 支持多種語言客戶端。RabbitMQ幾乎支持所有常用編程語言，包括 Java、.NET、Ruby 等等。
- 可視化管理界面。RabbitMQ提供了一個(gè)易用的用戶界面，使得用戶可以監(jiān)控和管理消息 Broker。
- 插件機(jī)制。RabbitMQ提供了許多插件，可以通過插件進(jìn)行擴(kuò)展，也可以編寫自己的插件。
為什么使用消息隊(duì)列
- 解耦，模塊間的通信使用消息隊(duì)列進(jìn)行，降低模塊之間的耦合度。
- 異步，模塊A處理完核心業(yè)務(wù)之后，發(fā)消息給模塊B，就可以直接返回給客戶端，提高性能。
- 削峰，短時(shí)間內(nèi)有大量請(qǐng)求，可以通過消息隊(duì)列來逐個(gè)處理，防止系統(tǒng)宕機(jī)。
Hello World
首先，安裝RabbitMQ，可直接使用docker安裝一個(gè)。
docker 地址：https://hub.docker.com/_/rabbitmq/tags
# 拉取鏡像
docker pull rabbitmq:management
# 啟動(dòng)容器
docker run -id --hostname my-rabbit \
--name rabbitmq \
-p 15672:15672 \
-p 5672:5672 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
rabbitmq:management
然后在網(wǎng)頁訪問：http://服務(wù)器的ip:15672，輸入賬號(hào)密碼。記得開放端口
看到此頁面代表安裝成功
準(zhǔn)備就緒，先寫個(gè)Hello World
導(dǎo)入pom依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
準(zhǔn)備工具類
public class ConnectionUtils {
public static Connection getConnection(){
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("1.15.88.28");
factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
try {
return factory.newConnection();
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
return null;
}
}生產(chǎn)者
public class Producer {
// 聲明隊(duì)列的名字
private static final String QUEUE_NAME = "queue_helloworld_1";
public static void main(String[] args) throws IOException, TimeoutException {
// 1. 獲取連接
Connection connection = ConnectionUtils.getConnection();
// 2. 創(chuàng)建數(shù)據(jù)傳輸通道
Channel channel = connection.createChannel();
// 3. 聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4. 發(fā)送數(shù)據(jù)到隊(duì)列
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, "第一個(gè)隊(duì)列消息...".getBytes());
// 5. 關(guān)閉通道
channel.close();
// 6. 關(guān)閉連接
connection.close();
}
}消費(fèi)者
public class Consumer {
// 聲明隊(duì)列的名字
private static final String QUEUE_NAME = "queue_helloworld_1";
public static void main(String[] args) throws IOException, TimeoutException {
// 1. 獲取連接
Connection connection = ConnectionUtils.getConnection();
// 2. 創(chuàng)建通道
Channel channel = connection.createChannel();
// 3. 聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4. 聲明消費(fèi)者
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接收到的消息是：" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 5. 綁定消費(fèi)者
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
}
}測試
- 先啟動(dòng)消費(fèi)者
- 再啟動(dòng)生產(chǎn)者
- 可以看到消費(fèi)者的控制臺(tái)打印出 生產(chǎn)者傳遞的消息
小結(jié)
本文到這里就結(jié)束了，簡單介紹了一下RabbitMQ是什么，以及RabbitMQ的安裝。
