🍅程式員小王的博客：程式員小王的博客
🍅 歡迎點贊 👍 收藏 ?留言 📝
🍅 如有編輯錯誤聯系作者，如果有比較好的文章歡迎分享給我，我會取其精華去其糟粕
🍅java自學的學習路線：java自學的學習路線

📒 小提示：RabbitMQ通過三篇博客進行記錄
 1. RabbitMQ安裝以及訊息模型使用攻略
 2. SpringBoot中使用RabbitMQ
 3. MQ的應用場景
 其中SpringBoot中使用RabbitMQ，MQ的應用場景將在明日發布

RabbitMQ(Rabbit Message queue)=訊息中間件，是一種理念（規范），RabbitMQ是其中的一個落地產品

一、MQ引言

1、什么是MQ？

MQ(Message Quene) : 翻譯為 訊息佇列,通過典型的 生產者和消費者模型,生產者不斷向訊息佇列中生產訊息，消費者不斷的從佇列中獲取訊息，因為訊息的生產和消費都是異步的，而且只關心訊息的發送和接收，沒有業務邏輯的侵入,輕松的實作系統間解耦，別名為 訊息中間件 通過利用高效可靠的訊息傳遞機制進行平臺無關的資料交流，并基于資料通信來進行分布式系統的集成，

訊息佇列：訊息(資料），佇列（排隊）先進先出

例：訂單系統（生產者），將數量這個資料給訊息中間件，庫存系統（消費者），消費訊息中間件的資料

2、MQ有哪些？

當今市面上有很多主流的訊息中間件，如老牌的ActiveMQ、RabbitMQ，炙手可熱的Kafka（大資料，高效，安全性低），阿里巴巴自主開發RocketMQ（OPPO 用的RocketMQ）等，

3、不同MQ的特點

# 1. ActiveMQ  JMS  java13種規范(介面)之一（jdbc）
    ActiveMQ 是Apache出品，最流行的，能力強勁的開源訊息總線，
    它是一個完全支持JMS規范的的訊息中間件，豐富的API,
    多種集群架構模式讓ActiveMQ在業界成為老牌的訊息中間件,在中小型企業頗受歡迎!

# 2. Kafka
    Kafka是LinkedIn開源的分布式發布-訂閱訊息系統，
    目前歸屬于Apache頂級專案，Kafka主要特點是基于Pull的模式來處理訊息消費，
    追求高吞吐量，一開始的目的就是用于日志收集和傳輸，
    0.8版本開始支持復制，不支持事務，對訊息的重復、丟失、錯誤沒有嚴格要求，
    適合產生大量資料的互聯網服務的資料收集業務，

# 3. RocketMQ
    RocketMQ是阿里開源的訊息中間件，
    它是純Java開發，具有高吞吐量、高可用性、適合大規模分布式系統應用的特點，
    RocketMQ思路起源于Kafka，
    但并不是Kafka的一個Copy，它對訊息的可靠傳輸及事務性做了優化，
    目前在阿里集團被廣泛應用于交易、充值、流計算、訊息推送、日志流式處理、binglog分發等場景，

# 4. RabbitMQ (安全)
    RabbitMQ是使用Erlang語言開發的開源訊息佇列系統，
    基于AMQP協議來實作，AMQP的主要特征是面向訊息、佇列、路由（包括點對點和發布/訂閱）、可靠性、安全，
    AMQP協議更多用在企業系統內對資料一致性、穩定性和可靠性要求很高的場景，對性能和吞吐量的要求還在其次，

RabbitMQ比Kafka可靠，Kafka更適合IO高吞吐的處理，一般應用在大資料日志處理或對實時性（少量延遲），可靠性（少量丟資料）要求稍低的場景使用，比如ELK日志收集，

二、RabbitMQ的引言

1、RabbitMQ官網

官網: https://www.rabbitmq.com/

官方教程: https://www.rabbitmq.com/#getstarted

基于AMQP協議，erlang語言開發，是部署最廣泛的開源訊息中間件,是最受歡迎的開源訊息中間件之一，

2、AMQP協議

AMQP（advanced message queuing protocol）先進的訊息排隊協議`在2003年時被提出，最早用于解決金融領域不同平臺之間的訊息傳遞互動問題，顧名思義，AMQP是一種協議，更準確的說是一種binary wire-level protocol（鏈接協議），這是其和JMS的本質差別，AMQP不從API層進行限定，而是直接定義網路交換的資料格式，這使得實作了AMQP的provider天然性就是跨平臺的，以下是AMQP協議模型:

Piblisher Application:生產者
Exchange:交換機
Virtual host:虛擬機
Message Queue:訊息佇列
Conbsumer Application:消費者

3、RabbitMQ 的安裝

（1）下載

官網下載地址: https://www.rabbitmq.com/download.html

（2）下載的安裝包

注意:這里的安裝包是centos7安裝的包

（3）安裝步驟

# 1.將rabbitmq安裝包上傳到linux系統中

  erlang-22.0.7-1.el7.x86_64.rpm  #erlang環境
  socat-1.7.3.2-2.el7.x86_64.rpm  #記憶體管理依賴
  rabbitmq-server-3.7.18-1.el7.noarch.rpm

# 2.安裝Erlang依賴包
  rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm
  rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

# 3.安裝RabbitMQ安裝包(需要聯網)
  yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm
  
    注意:默認安裝完成后組態檔模板在:
    /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example目錄中,
    需要將組態檔復制到/etc/rabbitmq/目錄中,并修改名稱為rabbitmq.config

# 4.復制組態檔
  cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

# 5.查看組態檔位置
  ls /etc/rabbitmq/rabbitmq.config

# 6.修改組態檔(參見下圖:)  ：61 查看61行
  vim /etc/rabbitmq/rabbitmq.config

# 7.執行如下命令,啟動rabbitmq中的插件管理
  rabbitmq-plugins enable rabbitmq_management
  
  出現如下說明:
    Enabling plugins on node rabbit@localhost:
    rabbitmq_management
    The following plugins have been configured:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    Applying plugin configuration to rabbit@localhost...
    The following plugins have been enabled:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch

    set 3 plugins.
    Offline change; changes will take effect at broker restart.

# 8.啟動RabbitMQ的服務
  systemctl start rabbitmq-server
  systemctl restart rabbitmq-server
  systemctl stop rabbitmq-server

注意：如果啟動出現問題  
Job for rabbitmq-server.service failed because the control process exited with error code  
需要修改主機名
    查看本機主機名 
        hostname 
    修改主機名
      hostnamectl set-hostname  主機名

# 10.關閉防火墻服務
systemctl stop firewalld   
systemctl disable firewalld
      Removed symlink /etc/systemd/system/multi-user.target.wants/firewalld.service.
      Removed symlink /etc/systemd/system/dbus-org.fedoraproject.FirewallD1.service.

# 11.訪問web管理界面（ 192.168.5.128是我自己的ip地址）
http:// 192.168.5.128:15672/

# 12.登錄管理界面 默認的
  username:  guest
  password:  guest

關閉RabbitMQ之后

systemctl stop rabbitmq-server

前端界面

4、RabbitMQ 配置

# 1.服務啟動相關
  systemctl start|restart|stop|status rabbitmq-server

# 2.管理命令列  用來在不使用web管理界面情況下命令操作RabbitMQ
  rabbitmqctl  help  可以查看更多命令

# 3.插件管理命令列
  rabbitmq-plugins enable|list|disable

5、web管理界面介紹

（1）overview概覽

connections：無論生產者還是消費者，都需要與RabbitMQ建立連接后才可以完成訊息的生產和消費，在這里可以查看連接情況
channels：通道，建立連接后，會形成通道，訊息的投遞獲取依賴通道，
Exchanges：交換機，用來實作訊息的路由
Queues：佇列，即訊息佇列，訊息存放在佇列中，等待消費，消費后被移除佇列，

（2）設定用戶名和密碼

上面的Tags選項，其實是指定用戶的角色，可選的有以下幾個：

超級管理員(administrator)
可登陸管理控制臺，可查看所有的資訊，并且可以對用戶，策略(policy)進行操作，
監控者(monitoring)
可登陸管理控制臺，同時可以查看rabbitmq節點的相關資訊(行程數，記憶體使用情況，磁盤使用情況等)
策略制定者(policymaker)
可登陸管理控制臺, 同時可以對policy進行管理，但無法查看節點的相關資訊(上圖紅框標識的部分)，
普通管理者(management)
僅可登陸管理控制臺，無法看到節點資訊，也無法對策略進行管理，
其他
無法登陸管理控制臺，通常就是普通的生產者和消費者，

（3）新建虛擬主機

虛擬主機：為了讓各個用戶可以互不干擾的作業，RabbitMQ添加了虛擬主機（Virtual Hosts）的概念，其實就是一個獨立的訪問路徑，不同用戶使用不同路徑，各自有自己的佇列、交換機，互相不會影響，

（4）系結虛擬主機和用戶

創建好虛擬主機，我們還要給用戶添加訪問權限：

點擊添加好的虛擬主機：

進入虛擬機設定界面:

設定成功后

三、RabbitMQ 的第一個程式

1、AMQP協議

2、RabbitMQ支持的訊息模型

3、引入依賴

RabitMQ的依賴

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.2</version>
</dependency>

SpringBoot環境的依賴

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.5.RELEASE</version>
  </parent>
  
<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
  </dependency>
  <!--RabitMQ-->
  <dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.2</version>
  </dependency>

</dependencies>

4、第一種模型 (hello World)

六種模型跟著官網學，如果看不懂因為，使用谷歌瀏覽器翻譯

在上圖的模型中，有以下概念：

P(Provider)：生產者，也就是要發送訊息的程式
C(Consumer)：消費者：訊息的接收者，會一直等待訊息到來，
queue：訊息佇列，圖中紅色部分，類似一個郵箱，可以快取訊息；生產者向其中投遞訊息，消費者從其中取出訊息，

（1）鏈接RabbitMQ

  @Test
    public void TestConnection() throws IOException, TimeoutException {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();
        System.out.println(connection);
        connection.close();
    }

（2)獲取通道channel

public class TestHello {
    //    生產者
    @Test
    public void provider() throws IOException, TimeoutException {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();
        System.out.println(connection);

        //獲取通道
        Channel channel = connection.createChannel();
        System.out.println(channel);

        channel.close();
        connection.close();
    }

（3)開發生產者

    //    生產者
    @Test
    public void provider() throws IOException, TimeoutException {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();
        System.out.println(connection);

        //獲取通道
        Channel channel = connection.createChannel();
        System.out.println(channel);
        /**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */
        channel.queueDeclare("HelloQueues",false,false,false,null);

        /**發布訊息
         * @param exchange 訊息交換機名稱,空字串將使用直接交換器模式，發送到默認的Exchange=amq.direct，此狀態下，RoutingKey默認和Queue名稱相同
         * @param queueName 佇列名稱
         * @param BasicProperties  設定訊息持久化：MessageProperties.PERSISTENT_TEXT_PLAIN是持久化；MessageProperties.TEXT_PLAIN是非持久化，
         * @param body 訊息物件轉換的byte[]
         */
        channel.basicPublish("","HelloQueues",null,"hello 2022".getBytes());

        //關閉連接
        channel.close();
        connection.close();
    }

注意：如果執行報錯 連接超時例外 需要修改linux系統中 /etc/hosts檔案

  vim /etc/hosts

    在127.0.0.1  此行最后加上主機名

（4）開發消費者

 //    消費者
    public static void main(String[] args) throws Exception {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();

        //獲取通道
        Channel channel = connection.createChannel();

        /**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */
        channel.queueDeclare("HelloQueues", true, false, true, null);

        /**消費者消費訊息
         * @param queue 佇列名稱
         * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
         * @param callback 回呼方法類，一般為自己的Consumer類
         */
        channel.basicConsume("HelloQueues", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                java.lang.String message = new String(body);

                System.out.println("消費者獲取訊息： " + message);
            }
        });
    }

rabbitmq不支持更改現有名稱的佇列，如javaQueues,直接改為false是不可以的，也就是說rabbitmq不支持重新定義一個已存在的佇列，

（5）設定佇列長度

HashMap<String, Object> map = new HashMap<>();
map.put("x-max-length",5);  //設定佇列最大長度
//可以通過為x-overflow佇列宣告引數提供字串值來設定溢位行為 ，
//可能的值為drop-head（默認）或 reject-publish
map.put("x-overflow","reject-publish");

/**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */
channel.queueDeclare("javaQueues",true,false,true,map);

案例

/**
 * @author 王恒杰
 * @date 2022/1/21 14:24
 * @Description:
 */
@RunWith(SpringRunner.class)
//指定測驗類啟動時入口類的位置  //等價于指定工廠的位置
@SpringBootTest(classes = RabbitMQApplication.class)
public class TestHello {
    //    生產者
    //    生產者
    @Test
    public void provider() throws IOException, TimeoutException {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();
        System.out.println(connection);

        //獲取通道
        Channel channel = connection.createChannel();
        System.out.println(channel);
        /**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */

        HashMap<String, Object> map = new HashMap<>();
        map.put("x-max-length",5);  //設定佇列最大長度
        //可以通過為x-overflow佇列宣告引數提供字串值來設定溢位行為 ，
        //可能的值為drop-head（默認）或 reject-publish
        //map.put("x-overflow","reject-publish");
        //4、設定佇列
        channel.queueDeclare("helloQueues",true,false,true,map);

        /**發布訊息
         * @param exchange 訊息交換機名稱,空字串將使用直接交換器模式，發送到默認的Exchange=amq.direct，此狀態下，RoutingKey默認和Queue名稱相同
         * @param queueName 佇列名稱
         * @param BasicProperties  設定訊息持久化：MessageProperties.PERSISTENT_TEXT_PLAIN是持久化；MessageProperties.TEXT_PLAIN是非持久化，
         * @param body 訊息物件轉換的byte[]
         */
        for (int i = 1; i < 11; i++) {
            String s = "hello"+i;
            //5、發布 生產訊息
            channel.basicPublish("","helloQueues",MessageProperties.PERSISTENT_TEXT_PLAIN,s.getBytes());
        }


        //關閉連接
        channel.close();
        connection.close();
    }



    //    消費者
    public static void main(String[] args) throws Exception {

        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");

        //獲取連接
        Connection connection = factory.newConnection();

        //獲取通道
        Channel channel = connection.createChannel();

        /**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */

        HashMap<String, Object> map = new HashMap<>();
        map.put("x-max-length",5);  //設定佇列最大長度
        //可以通過為x-overflow佇列宣告引數提供字串值來設定溢位行為 ，
        //可能的值為drop-head（默認）或 reject-publish
        //map.put("x-overflow","reject-publish");

        /**設定佇列引數
         * @param queue 佇列名稱  如果這個佇列不存在，將會被創建
         * @param durable 持久性：用來定義佇列是否要持久化  true:持久化  false:不持久化
         * @param exclusive 是否只能由創建者使用，其他連接不能使用， true:獨占佇列  false:不獨占佇列
         * @param autoDelete 是否自動洗掉（沒有連接自動洗掉） true:自動洗掉   false:不自動洗掉
         * @param arguments 佇列的其他屬性(構造引數)
         */
        channel.queueDeclare("helloQueues", true, false, true, map);

        /**消費者消費訊息
         * @param queue 佇列名稱
         * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
         * @param callback 回呼方法類，一般為自己的Consumer類
         */
        channel.basicConsume("helloQueues", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                java.lang.String message = new String(body);

                System.out.println("消費者獲取訊息： " + message);
            }
        });
    }
}

5、封裝RabbitMQ工具類

（1）RabbitMQ工具類

**
 * @author 王恒杰
 * @date 2022/1/21 16:10
 * @Description:
 */
public class RabbitMQUtil {
    private static ConnectionFactory factory = null;

   static{
        //創建連接工廠
        factory = new ConnectionFactory();
        //設定rabbitmq服務器IP地址
        factory.setHost("192.168.5.128");
        //設定rabbitmq服務器連接埠
        factory.setPort(5672);
        //設定rabbitmq服務器虛擬主機
        factory.setVirtualHost("yingxue");
        //設定rabbitmq服務器用戶名
        factory.setUsername("whj");
        //設定rabbitmq服務器密碼
        factory.setPassword("563135");
    }

    //獲取連接
    public static Connection getConnection(){
        Connection connection=null;
        try {
           connection= factory.newConnection();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        return  connection;
    }


    //關閉資源
    public static void close(Channel channel , Connection connection){
        try {
            if(channel!=null) {
                channel.close();
            }
            if(connection!=null) {
                connection.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

（2）生成者

/**
 * @author 王恒杰
 * @date 2022/1/22 15:48
 * @Description: 生產者
 */
public class Provider {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定佇列
        channel.queueDeclare("Hello_Util", true, false, true, null);
        //產生發送資料  Publish出版
        channel.basicPublish("", "Hello_Util", null, "hello王恒杰".getBytes());
        //關閉資源
        RabbitMQUtil.close(channel, connection);
    }
}

（3）消費者

/**
 * @author 王恒杰
 * @date 2022/1/22 15:55
 * @Description:
 */
public class Consumer {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定佇列
        channel.queueDeclare("Hello_Util", true, false, true, null);
        //消費者 消費資訊
        channel.basicConsume("Hello_Util", true,
                //匿名內部類
                new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        java.lang.String message = new String(body);
                        System.out.println("消費者獲取訊息： " + message);
                    }
                });

    }
}

5、第二種模型(work queue)

Work queues，也被稱為（Task queues），任務模型，當訊息處理比較耗時的時候，可能生產訊息的速度會遠遠大于訊息的消費速度，長此以往，訊息就會堆積越來越多，無法及時處理，此時就可以使用work 模型：讓多個消費者系結到一個佇列，共同消費佇列中的訊息，佇列中的訊息一旦消費，就會消失，因此任務是不會被重復執行的，

角色：

P：生產者：任務的發布者
C1：消費者-1，領取任務并且完成任務，假設完成速度較慢
C2：消費者-2：領取任務并完成任務，假設完成速度快

（1）開發生產者

/**
 * @author 王恒杰
 * @date 2022/1/22 19:47
 * @Description:
 */
public class Provider {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        // //設定佇列
        channel.queueDeclare("Work_王恒杰",true,false,true,null);
        //產生發送資料  Publish出版
        /**
         * @param exchange 訊息交換機名稱,空字串將使用直接交換器模式，發送到默認的Exchange=amq.direct，此狀態下，RoutingKey默認和Queue名稱相同
         * @param queueName 佇列名稱
         * @param BasicProperties  設定訊息持久化：MessageProperties.PERSISTENT_TEXT_PLAIN是持久化；MessageProperties.TEXT_PLAIN是非持久化，
         * @param body 訊息物件轉換的byte[]
         */
        for (int i = 0; i < 21; i++) {
            String message="Work_王恒杰"+i;
            channel.basicPublish("","Work_王恒杰",null,message.getBytes());
        }
        //關閉鏈接
        RabbitMQUtil.close(channel,connection);
    }
}

（2）開發消費者-1

/**
 * @author 王恒杰
 * @date 2022/1/22 19:47
 * @Description:
 */
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定佇列
        channel.queueDeclare("Work_王恒杰", true, false, true, null);
        //獲取訊息
        /**消費者消費訊息
         * @param queue 佇列名稱
         * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
         * @param callback 回呼方法類，一般為自己的Consumer類
         */
        channel.basicConsume("Work_王恒杰",true,
                //匿名內部類
                new DefaultConsumer(channel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        java.lang.String message = new String(body);
                        System.out.println("消費者1獲取訊息： " + message);
                    }
                }
                );
    }
}

（3）開發消費者-2

/**
 * @author 王恒杰
 * @date 2022/1/22 19:47
 * @Description:
 */
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定佇列
        channel.queueDeclare("Work_王恒杰", true, false, true, null);
        //獲取訊息
        /**消費者消費訊息
         * @param queue 佇列名稱
         * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
         * @param callback 回呼方法類，一般為自己的Consumer類
         */
        channel.basicConsume("Work_王恒杰",true,
                //匿名內部類
                new DefaultConsumer(channel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        java.lang.String message = new String(body);
                        System.out.println("消費者1獲取訊息： " + message);
                    }
                }
                );
    }
}

（4)測驗結果

總結:默認情況下，RabbitMQ將按順序將每個訊息發送給下一個使用者，平均而言，每個消費者都會收到相同數量的訊息，這種分發訊息的方式稱為回圈，

（5)平均分配出現的問題

讓消費者1，睡眠1000毫秒再消費

/**
 * @author 王恒杰
 * @date 2022/1/22 19:47
 * @Description:
 */
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定佇列
        channel.queueDeclare("Work_王恒杰", true, false, true, null);
        //獲取訊息
        /**消費者消費訊息
         * @param queue 佇列名稱
         * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
         * @param callback 回呼方法類，一般為自己的Consumer類
         */
        channel.basicConsume("Work_王恒杰",true,
                //匿名內部類
                new DefaultConsumer(channel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        java.lang.String message = new String(body);
                        //列印訊息前睡眠
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("消費者1獲取訊息： " + message);
                    }
                }
                );
    }
}

（6）訊息自動確認機制

完成一項任務可能需要幾秒鐘，您可能想知道，如果其中一個使用者開始一項漫長的任務并區域分完成而死掉，會發生什么情況，使用我們當前的代碼，RabbitMQ一旦向消費者傳遞了一條訊息，便立即將其標記為洗掉，在這種情況下，如果您殺死一個工人，我們將丟失正在處理的訊息，我們還將丟失所有發送給該特定作業人員但尚未處理的訊息，

但是我們不想丟失任何任務，如果一個工人死亡，我們希望將任務交付給另一個工人，

/**消費者消費訊息
     * @param queue 佇列名稱
     * @param autoAck 是否自動應答，false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條訊息消費失敗，繼續分發給其他consumer，
     * @param callback 回呼方法類，一般為自己的Consumer類
     */
channel.basicQos(1);//一次只接受一條未確認的訊息

//引數2:關閉自動確認訊息
channel.basicConsume("hello",false,new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("消費者1: "+new String(body));
    channel.basicAck(envelope.getDeliveryTag(),false);//手動確認訊息
  }
});

案例：讓消費者1，睡眠1000毫秒再消費，消費者2不睡眠

生產者

消費者1
在這里插入圖片描述

消費者2

測驗結果

6、第三種模型(fanout)

fanout 扇出也稱為廣播

在廣播模式下，訊息發送流程是這樣的：

可以有多個消費者
每個消費者有自己的queue（佇列）
每個佇列都要系結到Exchange（交換機）
生產者發送的訊息，只能發送到交換機，交換機來決定要發給哪個佇列，生產者無法決定，
交換機把訊息發送給系結過的所有佇列
佇列的消費者都能拿到訊息，實作一條訊息被多個消費者消費

(1). 開發生產者

@author 王恒杰
 * @date 2022/1/22 20:54
 * @Description:
 */
public class Provider{
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定交換機引數                   交換機名     交換機型別    持久化   自刪        其他
        channel.exchangeDeclare("fanout_王恒杰","fanout",true,true,null);
        //發布訊息                      交換機名  路由key   基本屬性        訊息內容
        channel.basicPublish("fanout_王恒杰","",null,"fonout_王恒杰，，，，".getBytes());

        //關閉鏈接
        RabbitMQUtil.close(channel, connection);

        System.out.println("生成資訊完成！");
    }
}

(2).開發消費者-1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //廣播 一條訊息多個消費者同時消費
        //設定交換機引數                交換機名     交換機型別     持久化         自刪         其他
        channel.exchangeDeclare("fanout_王恒杰", "fanout", true, true, null);
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange  臨時佇列名  交換機名   路由key
        channel.queueBind(queue,"fanout_王恒杰","");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者1: "+new String(body));
            }
        });
    }
}

(3). 開發消費者-2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //廣播 一條訊息多個消費者同時消費
        //設定交換機引數                交換機名     交換機型別     持久化         自刪         其他
        channel.exchangeDeclare("fanout_王恒杰", "fanout", true, true, null);
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange  臨時佇列名  交換機名   路由key
        channel.queueBind(queue,"fanout_王恒杰","");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者2: "+new String(body));
            }
        });
    }
}

(4).開發消費者-3

public class Consumer3 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //廣播 一條訊息多個消費者同時消費
        //設定交換機引數                交換機名     交換機型別     持久化         自刪         其他
        channel.exchangeDeclare("fanout_王恒杰", "fanout", true, true, null);
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange  臨時佇列名  交換機名   路由key
        channel.queueBind(queue,"fanout_王恒杰","");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者3: "+new String(body));
            }
        });
    }
}

(5). 測驗結果

7、Route 路由模型（訂閱模型-Direct(直連)）

Fanout模式中，一條訊息，會被所有訂閱的佇列都消費，但是，在某些場景下，我們希望不同的訊息被不同的佇列消費，這時就要用到Direct型別的Exchange，

在Direct模型下：

佇列與交換機的系結，不能是任意系結了，而是要指定一個RoutingKey（路由key）
訊息的發送方在向 Exchange發送訊息時，也必須指定訊息的 RoutingKey，
Exchange不再把訊息交給每一個系結的佇列，而是根據訊息的Routing Key進行判斷，只有佇列的Routingkey與訊息的 Routing key完全一致，才會接收到訊息

流程:

圖解：

P：生產者，向Exchange發送訊息，發送訊息時，會指定一個routing key，
X：Exchange（交換機），接收生產者的訊息，然后把訊息遞交給與routing key完全匹配的佇列
C1：消費者，其所在佇列指定了需要routing key 為 error 的訊息
C2：消費者，其所在佇列指定了需要routing key 為 info、error、warning 的訊息

(1). 開發生產者

public class Provider {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定交換機引數                   交換機名     交換機型別（Direct(直連)  靜態key）
        channel.exchangeDeclare("exchange_王恒杰","direct");
        //發布訊息                      交換機名  路由key   基本屬性        訊息內容
        channel.basicPublish("exchange_王恒杰","user",null,"wxchange_direct_王恒杰，，，，".getBytes());

        //關閉鏈接
        RabbitMQUtil.close(channel, connection);

        System.out.println("生成資訊完成！");
    }
}

(2).開發消費者-1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //廣播 一條訊息多個消費者同時消費
        //設定交換機引數                                     Direct(直連)
        channel.exchangeDeclare("exchange_王恒杰", "direct");
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange 引數1：臨時佇列名  引數2：交換機名 引數3：交換機key(category：類別)
        channel.queueBind(queue, "exchange_王恒杰", "category");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者1: " + new String(body));
            }
        });
    }
}

(3). 開發消費者-2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //廣播 一條訊息多個消費者同時消費
        //設定交換機引數                                     Direct(直連)
        channel.exchangeDeclare("exchange_王恒杰", "direct");
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange 引數1：臨時佇列名  引數2：交換機名(category：類別) 引數3：交換機key
        channel.queueBind(queue, "exchange_王恒杰", "user");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者1: " + new String(body));
            }
        });
    }
}

(4). 測驗結果

消費者1

消費者2

8、Topic 訂閱模型(動態路由模型)

Topic型別的Exchange與Direct相比，都是可以根據RoutingKey把訊息路由到不同的佇列，只不過Topic型別Exchange可以讓佇列在系結Routing key 的時候使用通配符！這種模型Routingkey 一般都是由一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符

  * (star) can substitute for exactly one word.    匹配不多不少恰好1個詞
  # (hash) can substitute for zero or more words.  匹配一個或多個詞
如:
  audit.*   只能匹配 audit.irs
  audit.#   可匹配audit.irs.corporate 或者 audit.irs 等

(1). 開發生產者

/**
 * @author 王恒杰
 * @date 2022/1/22 21:46
 * @Description:
 */
public class Provider {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        Channel channel = connection.createChannel();
        //設定交換機  引數1:交換機名稱 引數2:交換機型別 topic 使用動態路由(通配符方式)
        channel.exchangeDeclare("topic_王恒杰", "topic");
        //   動態路由key
        String routeKey = "user.save";
        //發布訊息 引數1：交換機名 引數2：路由key 引數3：是否持久化 引數4：訊息內容
        channel.basicPublish("topic_王恒杰", routeKey, null, ("這是路由的動態訂閱模型,route key:[" + routeKey + "]").getBytes());
        //關閉資源
        RabbitMQUtil.close(channel, connection);
        System.out.println("生產資訊完成....");
    }
}

(2).開發消費者-1

Routing Key中使用*通配符方式

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //設定交換機引數  引數1:交換機名稱 引數2:交換機型別 topic 使用動態路由(通配符方式)
        channel.exchangeDeclare("topic_王恒杰","topic");
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange 引數1：臨時佇列名  引數2：交換機名 引數3：動態路由key
        channel.queueBind(queue, "topic_王恒杰", "user.*");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者1: " + new String(body));
            }
        });
    }
}

(3). 開發消費者-2

Routing Key中使用#通配符方式

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        //獲取鏈接
        Connection connection = RabbitMQUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //設定交換機引數  引數1:交換機名稱 引數2:交換機型別 topic 使用動態路由(通配符方式)
        channel.exchangeDeclare("topic_王恒杰","topic");
        //創建臨時佇列
        String queue = channel.queueDeclare().getQueue();
        //將臨時佇列系結到exchange 引數1：臨時佇列名  引數2：交換機名 引數3：動態路由key
        channel.queueBind(queue, "topic_王恒杰", "user.#");
        //消費訊息           臨時佇列名  是否自動應答     callback
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者2: " + new String(body));
            }
        });
    }
}

(4). 測驗結果

消費者-1：Routing Key中使用*通配符方式

消費者-2：Routing Key中使用#通配符方式

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/419144.html

標籤：其他

上一篇：阿里云MaxComputer SQL學習之DML

下一篇：只有命名空間中的第一個元素回傳一個物件