在工作中經(jīng)常會(huì)用到消息隊(duì)列處理各種問(wèn)題，今天指北君帶領(lǐng)大家來(lái)學(xué)一個(gè)很常用到的技術(shù)-RabbitMQ；接下來(lái)還會(huì)有關(guān)于RabbitMQ的系列教程，對(duì)你有幫助的話(huà)記得關(guān)注哦~

RabbitMQ是什么

隊(duì)列 ：一種數(shù)據(jù)結(jié)構(gòu)，先進(jìn)先出。

消息隊(duì)列 ：簡(jiǎn)單的說(shuō)就是用來(lái)進(jìn)行消息傳輸?shù)年?duì)列。

消息中間件 ：簡(jiǎn)單的說(shuō)就是用來(lái)傳輸消息的中間載體，就是將你的信息發(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ā)消息策略有：簡(jiǎn)單模式、工作隊(duì)列模式、發(fā)布訂閱模式、路由模式、通配符模式。
• 支持集群。多臺(tái)RabbitMQ服務(wù)器可以組成一個(gè)集群，形成一個(gè)邏輯Broker。
• 多種協(xié)議。RabbitMQ支持多種消息隊(duì)列協(xié)議，比如 STOMP、MQTT 等等。
• 支持多種語(yǔ)言客戶(hù)端。RabbitMQ幾乎支持所有常用編程語(yǔ)言，包括 Java、.NET、Ruby 等等。
• 可視化管理界面。RabbitMQ提供了一個(gè)易用的用戶(hù)界面，使得用戶(hù)可以監(jiān)控和管理消息 Broker。
• 插件機(jī)制。RabbitMQ提供了許多插件，可以通過(guò)插件進(jìn)行擴(kuò)展，也可以編寫(xiě)自己的插件。

為什么使用消息隊(duì)列

1. 解耦 ，模塊間的通信使用消息隊(duì)列進(jìn)行，降低模塊之間的耦合度。
2. 異步 ，模塊A處理完核心業(yè)務(wù)之后，發(fā)消息給模塊B，就可以直接返回給客戶(hù)端，提高性能。
3. 削峰 ，短時(shí)間內(nèi)有大量請(qǐng)求，可以通過(guò)消息隊(duì)列來(lái)逐個(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)頁(yè)訪問(wèn)：http://服務(wù)器的ip:15672，輸入賬號(hào)密碼。記得開(kāi)放端口

看到此頁(yè)面代表安裝成功

準(zhǔn)備就緒，先寫(xiě)個(gè)Hello World

1. 導(dǎo)入pom依賴(lài)

   < dependency >
        < groupId >org.springframework.boot< /groupId >
        < artifactId >spring-boot-starter-amqp< /artifactId >
   < /dependency >
   

2. 準(zhǔn)備工具類(lèi)

   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;
       }
   }
   

3. 生產(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();
       }
   }
   

4. 消費(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);
       }
   }
   

5. 測(cè)試
   1. 先啟動(dòng)消費(fèi)者
   2. 再啟動(dòng)生產(chǎn)者
   3. 可以看到消費(fèi)者的控制臺(tái)打印出 生產(chǎn)者傳遞的消息
      

小結(jié)

本文到這里就結(jié)束了，簡(jiǎn)單介紹了一下RabbitMQ是什么，以及RabbitMQ的安裝；