應用場景：

? 異步處理

• ? 用戶注冊后，需要發送郵箱和手機驗證碼：

  ? 傳統的需要一個步驟一個步驟的完成

  

  ? 使用訊息佇列后可以在發送郵箱的同時發送驗證碼

  

• 應用解耦

  ? 將模塊與模塊之間的呼叫相互分離，不在是模塊之間的直接呼叫

  

流量削峰

? 場景：秒殺，搶購業務

? 因為流量過大，導致服務宕機，為解決這個問題，在前端加入訊息佇列

?

? 用戶發起秒殺請求，將請求加入訊息佇列中，超過佇列長度就拋棄，拋出秒殺失敗頁面，

RabbitMQ和Kafka相比：RabbitMQ更可靠 ，相比Kafka性能低，

Kafka性能高（高的離譜，壓RabbitMQ一頭）但是相對于來說沒有RabbitMQ可靠，根據場景選擇訊息佇列，

RabbitMQ各組件功能

RabbitMQ安裝與使用

想要安裝RabbitMQ，必須先安裝erlang語言環境，類似安裝tomcat，必須先安裝JDK

查看RabbitMQ匹配的Erlang版本：https://www.rabbitmq.com/which-erlang.html

RabbitMQ版本 支持的Erlang最低版本 支持的Erlang最高版本

RabbitMQ的安裝：

RabbitMQ高性能訊息佇列要安裝在Linux上而不是Windows上，

erlang下載：https://dl.bintray.com/rabbitmq-erlang/rpm/erlang

socat下載：http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm

RabbitMQ下載：https://www.rabbitmq.com/install-rpm.html#downloads

下載完成后 發送到Linux上，在Linux上進行安裝,下方安裝命令按照順序執行，

 [root@localhost opt]# rpm -ivh erlang-21.3.8.16-1.el7.x86_64.rpm

 [root@localhost opt]# rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm

 [root@localhost opt]# rpm -ivh rabbitmq-server-3.8.6-1.el7.noarch.rpm

RabbitMQ自帶有一個豐富的后臺管理頁面，需要我們自行啟動

[root@localhost opt]# rabbitmq-plugins enable rabbitmq_management

開啟RabbitMQ

[root@localhost opt]#	systemctl start rabbitmq-server.service

查看RabbitMQ狀態

[root@localhost opt]#	systemctl status rabbitmq-server.service

重啟RabbitMQ

[root@localhost opt]#	systemctl restart rabbitmq-server.service

停止運行RabbitMQ

[root@localhost opt]#	systemctl stop rabbitmq-server.service

查看RabbitMQ相關行程

[root@localhost opt]#	ps -ef | grep rabbitmq

訪問RabbitMQ自帶的后臺頁面，因為本地安裝的Centos自啟動有防火墻，我們要訪問Linux需要先關閉防火墻或者開放埠，

[root@localhost opt]#	systemctl stop firewalld

訪問頁面，在網址舒服Linux所在IP地址+埠號15672，

默認賬號密碼是guest ，guest，默認賬號密碼是不支持遠程訪問的，所以我們要新建一個用戶

添加賬號命令

[root@localhost opt]#	rabbitmqctl add_user 自定義用戶名 密碼

分配超級管理員角色

[root@localhost opt]#	rabbitmqctl set_user_tags 用戶名 administrator

設定角色權限

[root@localhost opt]#	rabbitmqctl set_permissions -p "/" 用戶名 ".*" ".*" ".*"

查看角色串列

[root@localhost opt]#	rabbitmqctl list_users

修改密碼命令

[root@localhost opt]#	rabbitmqctl change_password 用戶名  要改成的密碼

訪問成功頁面

RabbitMQ帶有三個埠

5672：RabbitMQ提供給編程語言客戶端鏈接的埠，

15672：RabbitMQ后臺管理界面埠號，

25672：RabibtMQ集群用的埠號，

Java與RabbitMQ連接使用

引入依賴

<dependencies>
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.3</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.25</version>
    <scope>compile</scope>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.9</version>
</dependency>
</dependencies>

配置Log4j日志檔案

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %m%n
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.File=rebbitmq.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %l %m%n
log4j.rootLogger=debug, stdout,file

在IDEA中創建Java與RabbitMQ連接的工具類，因為在創建連接的時候需要選擇VirtuaHost虛擬主機 剛開始安裝的RabbitMQ只有一個虛擬主機 **/**我們可以自定義創建虛擬主機，可以在后臺頁面中創建

創建ConnectionUtls類獲取連接

public static Connection getConnection() throws IOException, TimeoutException {
    //創建連接工廠
    ConnectionFactory factory = new ConnectionFactory();
    //設定連接埠號
    factory.setPort(5672);
    //設定主機地址
    factory.setHost("192.168.152.130");
    //設定連接RabbitMQ的哪個虛擬主機
    factory.setVirtualHost("/lyux");
    //設定賬號密碼
    factory.setUsername("lyx");
    factory.setPassword("123456");
    //創建連接
    Connection connection = factory.newConnection();
    return connection;
}

RabbitMQ模式

RabbitMQ一共有6個模式，先學習前五個模式，第六個模式是RPC遠程呼叫，

前五個模式可以分為兩大類：點對點模式、發布訂閱模式，

點對點模式：PSP（point to point）模式包含三個角色

• 發布者、訊息佇列、接收者

• 佇列中保存訊息，直到訊息被消費或者超時

• 每個訊息發送到訊息佇列中進行保存，接收者從訊息佇列中獲取訊息

• 特點：

  • ? 每個訊息只有一個消費者，被消費后就不在訊息佇列了

  • ? 發送者與接收者沒有依賴關系，發送者發送的訊息不管接收者接受與否都會放在訊息佇列中，

  • 接收者接受訊息后，需要向發送物件發送應答成功

  如果需要保證每條訊息都被接受就需要用到P2P模式

發布訂閱模式（publish/subscribe)包含三個角色

• 發布者、交換機、訂閱者

• 多個發布者將訊息發送到交換機，交換機將訊息分別發送到多個訂閱者

• 特點

  • 每個訊息可以有多個訂閱者

  • 發布者與訂閱者有時間上的依賴，也就是對于某一個訂閱者來說，只有訂閱了發布者，才能消費發布者的訊息

  • 為了消費發布者資訊，訂閱者需要保持運行狀態

  • 如果希望發送的一個訊息被多人使用可以采用這個模式，

點對點模式之簡單模式

創建訊息發送方

ppublic class Sender{
    public static void  main(String[] args){
        //利用創建好的ConnectionUtls類獲取連接
       Connection connection =  ConnectionUtils.getConnection();
        //創建信道 
        Channel channel = connection.createChannel();
      
//引數1 – 佇列的名稱
//引數2 - 如果我們宣告一個持久佇列，則為真（該佇列將在服務器重啟后繼續存在）
//引數3 – 如果我們宣告獨占佇列（僅限于此連接），則為 true
//引數4 – 如果我們宣告一個自動洗掉佇列，則為 true（服務器將在不再使用時將其洗掉）
//引數5  - 佇列的其他屬性（構造引數）
          //創建佇列
 		channel.queueDeclare("queue1",false,false,false,null);
        
//引數1 - 將訊息發布到的交換機
//引數2 – 佇列名
//引數3 - 如果要設定“強制”標志，則為 true
//引數4 - 訊息的其他屬性 - 路由標頭等
//引數5 - 訊息正文
        //發送資訊
        channel.basicPublish("","queue1",null,msg.getBytes(StandardCharsets.UTF_8));
        //關閉連接
        channel.close();
        connection.close();
    }
}

創建接收方

public class Accept{
    public static void main(String[] args){
       //持續保持連接，不然無法接受訊息
        while (true) {
        //獲取鏈接
         	Connection connection = ConnectionUtils.getConnection();
            //創建信道
         	Channel channel = connection.createChannel();
        	//創建獲取佇列后回呼的物件 
        	DefaultConsumer consumer = 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);
                }
            };
        //獲取佇列內容
         	channel.basicConsume("queue1", false, consumer); 
        } 
    }
}

初次創建 一定要先啟動 發送方，因為先啟動接收方的話 沒有找到相應的佇列會報錯，如果想要先啟動接收方的話，可以用下方的方法，

public class Accept{
    public static void main(String[] args){
       //持續保持連接，不然無法接受訊息
        while (true) {
        //獲取鏈接
         	Connection connection = ConnectionUtils.getConnection();
            //創建信道
         	Channel channel = connection.createChannel();
            //如果佇列沒有被創建，就創建佇列，如果已經被創建，就獲取佇列
             channel.queueDeclare("queue1",false,false,false,null);
        	//創建獲取佇列后回呼的物件 
        	DefaultConsumer consumer = 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);
                }
            };
        //獲取佇列內容
         	channel.basicConsume("queue1", false, consumer); 
        } 
    }
}

點對點模式之作業佇列模式

? 作業佇列模式就是發送方產生訊息過快，而接收方接收訊息慢，使訊息不能及時使用，這時我們就可以多有幾個接收方，

1、創建兩個接收方代碼相同

public class Accept{
        public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        //創建信道
        Channel channel = connection.createChannel();
        channel.queueDeclare("queue1", false, false, false, null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body));
                try {
                    //延遲時間不同，達到勞動能力不同的效果
                    //時間小的就是干得快的，時間慢的就是干得慢的，
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("queue1", false, consumer);
    }
}

先運行兩個接收方，在運行發送方，運行后我們發現并沒有達到干得快就干得多的效果，

官方原話如下

為了解決這個問題我們可以在接受方加入

int prefetchCount = 1;
channel.basicQos(prefetchCount);

成品如下

public class Accept {
        public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        //創建信道
        Channel channel = connection.createChannel();
        channel.queueDeclare("queue1", false, false, false, null);
        	//告訴RabbitMQ 不要向接收方一次發送多條訊息，只有確認一條后才發送另外一條，
            channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("queue1", false, consumer);
    }
}

此時我們就達到了能者多干的效果，

• 面試題：怎么避免訊息堆積？
  • ? 可以采用多個作業方監控發送方的訊息
  • ? 通過執行緒池 ，異步消費，

發布訂閱模式

簡單來說 就相當于抖音快手粉絲訂閱主播，主播發送視頻時，每一個粉絲都能收到，而不是只有一個人能收到訊息，

相對于點對點模式來說 多了一個交換機（路由），

創建發送方，

public class Sender {
    public static void main(String[] args) throws IOException, TimeoutException {

        String msg=  "你好,世界";
        //獲取連接
        Connection connection = ConnectionUtils.getConnection();
        //創建信道
        Channel channel = connection.createChannel();
        //創建路由(路由名字,路由型別)
        //路由型別可以在后臺管理頁面中查看
        //fanout
        channel.exchangeDeclare("yangmi","fanout");
        //發送資訊
        channel.basicPublish("yangmi","",null,msg.getBytes());
        channel.close();
        connection.close();
    }
}

創建兩個接收方，代碼基本相同，佇列名字不同

public class Accept {
    public static void main(String[] args) throws IOException, TimeoutException {
        //獲取鏈接
        Connection connection = ConnectionUtils.getConnection();
        //創建信道
        Channel channel = connection.createChannel();
        //創建佇列
        channel.queueDeclare("fensi2",false,false,false,null);
        //系結交換機
        channel.queueBind("fensi2","yangmi","");
        //創建接收訊息后的回呼類
        DefaultConsumer consumer = 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);
            }
        };
        //創建消費者
        channel.basicConsume("fensi2",false,consumer);
    }
}

發送方發送訊息，接收方接受訊息，我們發現兩個接收方能接受的是同一個訊息，

同一個訊息能被多個消費者消費，這就是發布訂閱模式