簡單的說就是用來傳輸消息的中間載體，就是將你的信息發(fā)送到接受方，它并不關(guān)心發(fā)送的數(shù)據(jù)是什么。RabbitMQ就是一個消息中間件。
大家好，我是指北君。
在工作中經(jīng)常會用到消息隊列處理各種問題，今天指北君帶領(lǐng)大家來學一個很常用到的技術(shù)-RabbitMQ；接下來還會有關(guān)于RabbitMQ的系列教程。
RabbitMQ是什么
隊列：一種數(shù)據(jù)結(jié)構(gòu)，先進先出。
消息隊列：簡單的說就是用來進行消息傳輸?shù)年犃小?/p>
消息中間件：簡單的說就是用來傳輸消息的中間載體，就是將你的信息發(fā)送到接受方，它并不關(guān)心發(fā)送的數(shù)據(jù)是什么。RabbitMQ就是一個消息中間件。
RabbitMQ的特點
- 可靠性。支持持久化，傳輸確認，發(fā)布確認等保證了MQ的可靠性。
- 靈活的分發(fā)消息策略。這應(yīng)該是RabbitMQ的一大特點。在消息進入MQ前由Exchange(交換機)進行路由消息。分發(fā)消息策略有：簡單模式、工作隊列模式、發(fā)布訂閱模式、路由模式、通配符模式。
- 支持集群。多臺RabbitMQ服務(wù)器可以組成一個集群，形成一個邏輯Broker。
- 多種協(xié)議。RabbitMQ支持多種消息隊列協(xié)議，比如 STOMP、MQTT 等等。
- 支持多種語言客戶端。RabbitMQ幾乎支持所有常用編程語言，包括 Java、.NET、Ruby 等等。
- 可視化管理界面。RabbitMQ提供了一個易用的用戶界面，使得用戶可以監(jiān)控和管理消息 Broker。
- 插件機制。RabbitMQ提供了許多插件，可以通過插件進行擴展，也可以編寫自己的插件。
為什么使用消息隊列
- 解耦，模塊間的通信使用消息隊列進行，降低模塊之間的耦合度。
- 異步，模塊A處理完核心業(yè)務(wù)之后，發(fā)消息給模塊B，就可以直接返回給客戶端，提高性能。
- 削峰，短時間內(nèi)有大量請求，可以通過消息隊列來逐個處理，防止系統(tǒng)宕機。
Hello World
首先，安裝RabbitMQ，可直接使用docker安裝一個。
docker 地址：https://hub.docker.com/_/rabbitmq/tags
# 拉取鏡像
docker pull rabbitmq:management
# 啟動容器
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，輸入賬號密碼。記得開放端口
看到此頁面代表安裝成功
準備就緒，先寫個Hello World
導入pom依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
準備工具類
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 {
// 聲明隊列的名字
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. 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4. 發(fā)送數(shù)據(jù)到隊列
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, "第一個隊列消息...".getBytes());
// 5. 關(guān)閉通道
channel.close();
// 6. 關(guān)閉連接
connection.close();
}
}消費者
public class Consumer {
// 聲明隊列的名字
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. 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4. 聲明消費者
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. 綁定消費者
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
}
}測試
- 先啟動消費者
- 再啟動生產(chǎn)者
- 可以看到消費者的控制臺打印出 生產(chǎn)者傳遞的消息
小結(jié)
本文到這里就結(jié)束了，簡單介紹了一下RabbitMQ是什么，以及RabbitMQ的安裝。
