03-RabbitMQ的工作模式

来源:https://www.cnblogs.com/lannntuuu/archive/2023/03/09/17199308.html
-Advertisement-
Play Games

RabbitMQ的工作模式 一、模式概述 **RabbitMQ提供了6種工作模式：**簡單模式、工作隊列模式、訂閱模式、路由模式、通配符模式、遠程調用模式 其中遠程調用模式（RPC）暫不作介紹。 官網對於模式介紹：https://www.rabbitmq.com/getstarted.html 二、 ...


RabbitMQ的工作模式

一、模式概述

RabbitMQ提供了6種工作模式：簡單模式、工作隊列模式、訂閱模式、路由模式、通配符模式、遠程調用模式

其中遠程調用模式（RPC）暫不作介紹。

官網對於模式介紹：https://www.rabbitmq.com/getstarted.html

RabbitMQ六種工作模式

二、簡單模式 HelloWorld

2.1、模式說明

簡單模式

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

  • P：生產者，也就是要發送消息的程式
  • C：消費者，消息的接收者，會一直等待消息到來
  • Queue：消息隊列，圖中紅色部分。類似一個郵箱，可以緩存消息；生產者向其中投遞消息，消費者從其中取出消息。

2.2、代碼舉例

需求：使用簡單模式完成消息傳遞

步驟：

  1. 創建工程（生產者、消費者）
  2. 分別添加依賴
  3. 編寫生產者發送消息
  4. 編寫消費者接收消息

1、創建工程

創建空項目RabbitMQ，然後在RabbitMQ下麵創建生產者rabbitmq-producer、消費者rabbitmq-consumer模塊（New Module - Build System選Maven）

簡單模式-創建生產者消費者工程

簡單模式-工程結構

2、添加依賴和編譯插件

這裡生產者和消費者是一樣的

maven-compiler-plugin 插件是一個 Maven 插件，用來編譯項目代碼。

自從3.0開始預設的編譯器是 javax.tools.JavaCompiler，用來編譯 Java 源碼；

如果你想強制插件使用 javac 編譯器，你必須配置插件的屬性 forceJavacCompilerUse；

還要註意，當前預設源（source）設置為 1.8，預設目標(target)設置為 1.8。獨立運行 Maven 和 JDK，可以通過 source 和 target 選項更改他們的預設值；

插件設置的各種參數信息請查看

http://maven.apache.org/plugins/maven-compiler-plugin/compile-mojo.html#forceJavacCompilerUse

<dependencies>
    <!--rabbitmq java 客戶端-->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>


<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

3、編寫生產者發送消息

public class Producer_HelloWorld {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        參數：
            1、queue：隊列名稱
            2、durable：是否持久化，當mq重啟之後，還在
            3、exclusive：
                * 是否獨占。只能有一個消費者監聽這隊列
                * 當Connection關閉時，是否刪除隊列
            4、autoDelete：是否自動化刪除。當沒有Consumer時，自動刪掉
            5、arguments：參數。
         */
        
        // 如果沒有一個名字叫 Hello_World 的隊列，則會創建該隊列，如果有則不會創建
        channel.queueDeclare("Hello_World", true, false, false, null);
        
        // 6、發送消息
        /*
        basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)
        參數：
            1、exchange：交換機名稱。簡單模式下交換機會使用預設的 ""
            2、routingKey：路由名稱
            3、props：配置信息
            4、body：發送消息數據
         */
        String body = "Hello RabbitMQ~~~";
        
        channel.basicPublish("", "Hello_World", null, body.getBytes());

        // 7、釋放資源
        channel.close();
        connection.close();
    }
}

4、編寫消費者接收消息

public class Consumer_HelloWorld {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        參數：
            1、queue：隊列名稱
            2、durable：是否持久化，當mq重啟之後，還在
            3、exclusive：
                * 是否獨占。只能有一個消費者監聽這隊列
                * 當Connection關閉時，是否刪除隊列
            4、autoDelete：是否自動化刪除。當沒有Consumer時，自動刪掉
            5、arguments：參數。
         */
        
        // 如果沒有一個名字叫 Hello_World 的隊列，則會創建該隊列，如果有則不會創建
//        channel.queueDeclare("Hello_World", true, false, false, null);
        
        // 6、接收消息
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        參數：
            1、queue：隊列名稱
            2、autoAck：是否自動確認
            2、callback：回調對象
         */

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /*
                回調方法，當收到消息之後，會自動執行該方法
                
                1、consumerTag：標識
                2、envelope：獲取一些信息，交換機，路由key...
                3、properties：配置信息
                4、body：數據
             */

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag：" + consumerTag);
                System.out.println("Exchange：" + envelope.getExchange());
                System.out.println("RoutingKey：" + envelope.getRoutingKey());
                System.out.println("properties：" + properties);
                System.out.println("body：" + new String(body));
            }
        };
        
        channel.basicConsume("Hello_World", true, consumer);

        // 7、釋放資源 不需要
    }
}

5、結果

  1. 啟動生產者，在虛擬機中創建隊列，像虛擬機中指定隊列發送消息
  2. 啟動消費者，與虛擬機建立連接，監聽指定隊列，接收消息

簡單模式-結果1

簡單模式-結果2

三、工作隊列模式（Work Queues）

3.1、模式說明

工作隊列模式

  • Work Queues：與簡單模式相比，多了一個或多個消費者，多個消費者共同消費同一個隊列中的消息。
  • 應用場景：對於任務過重或任務較多情況使用工作隊列可以提高任務處理的速度。

3.2、代碼舉例

Work Queues與簡單模式的代碼幾乎是一樣的。可以完全複製，並多複製一個消費者進行多個消費者同時對消費者的測試。

1、生產者

public class Producer_WorkQueues {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        channel.queueDeclare("Work_Queues", true, false, false, null);
        
        // 6、發送消息
        for (int i = 0; i < 10; i++) {
            String body = i + "Hello RabbitMQ~~~";

            channel.basicPublish("", "Work_Queues", null, body.getBytes());
        }

        // 7、釋放資源
        channel.close();
        connection.close();
    }
}

2、消費者1/消費者2

public class Consumer_WorkQueues1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();
        
        // 5、創建隊列
        channel.queueDeclare("Work_Queues", true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /*
                回調方法，當收到消息之後，會自動執行該方法
                
                1、consumerTag：標識
                2、envelope：獲取一些信息，交換機，路由key...
                3、properties：配置信息
                4、body：數據
             */

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };
        
        channel.basicConsume("Work_Queues", true, consumer);

        // 7、釋放資源 不需要
    }
}

3、結果

先啟動生產者再啟動消費者會導致第一個啟動的消費者把消息全消費了。

所以先啟動兩個消費者，再啟動生產者，這樣就可以看到兩個消費者是輪詢著消費。

當然也可以把發送消息單獨寫一個類出來，或者延時執行。

工作隊列模式-結果1

工作隊列模式-結果2

工作隊列模式-結果3

4、小結

  1. 在一個隊列中如果有多個消費者，那麼對於同一個消息來說消費者之間是競爭關係。
  2. Work Queues對於任務過重或任務較多情況使用工作隊列可以提高任務處理的速度。例如：簡訊服務部署多個，只需要有一個節點成功發送即可。

四、訂閱模式 （Publish/Subscribe）

4.1、模式說明

訂閱模式

在訂閱模型中，多了一個Exchange角色，而且過程略有變化：

  • P：生產者，也就是要發送消息的程式，但是不再直接發送到隊列中，而是發給X（交換機）

  • C：消費者，消息的接收者，會一直等待消息到來

  • Queue：消息隊列，接收消息、緩存消息

  • X：Exchange（交換機），一方面，接收生產者發送的消息。另一方面，直到如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。

    Exchange有常見以下3種類型：

    ​ Fanout：廣播，將消息交給所有綁定到交換機的隊列

    ​ Direct：定向，把消息交給符合Routing Key的隊列

    ​ Topic：通配符，把消息交給符合Routing Pattern（路由模式）的隊列

Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那麼消息會丟失！

4.2、代碼舉例

1、生產者

public class Producer_PubSub {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建交換機
        /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        參數：
            1、exchange：交換機名稱
            2、type：交換機類型    
                DIRECT("direct")：定向
                FANOUT("fanout")：扇形（廣播），發生消息到每一個與之綁定隊列。
                TOPIC("topic")：通配符的方式
                HEADERS("headers")：參數匹配
            3、durable：是否持久化
            4、autoDelete：自動刪除
            5、internal：內部使用。一般false
            6、arguments：參數
            
         */
        String exchangeName = "test_fanout";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);
        
        // 6、創建隊列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、綁定隊列和交換機
        /*
        queueBind(String queue, String exchange, String routingKey)
        參數：
            1、queue：隊列名稱
            2、exchange：交換機名稱
            3、routingKey：路由鍵，綁定規則
                如果交換機的類型為fanout，routingkey設置為""
         */
        channel.queueBind(queue1Name, exchangeName, "");
        channel.queueBind(queue2Name, exchangeName, "");
        
        // 8、發送消息
        String body = "日誌信息：張三調用了findAll方法...日誌級別：info...";
        
        channel.basicPublish(exchangeName, "", null, body.getBytes());
        // 9、釋放資源
        channel.close();
        connection.close();
    }
}

2、消費者

消費者1

public class Consumer_PubSub1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        String queue1Name = "test_fanout_queue1";
        channel.queueDeclare(queue1Name, true, false, false, null);
        
        // 6、接收消息

        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息列印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

消費者2

public class Consumer_PubSub2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue2Name, true, false, false, null);

        // 6、接收消息

        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息保存到資料庫.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

3、結果

先啟動生產者，會創建廣播模式的交換機和隊列，然後綁定交換機和隊列。之後發送消息，通過交換機廣播發送到兩個綁定的隊列中。

兩個消費者收到消息然後分別執行自己的後續邏輯。

訂閱模式-結果03

訂閱模式-結果04

訂閱模式-結果01

圖片/訂閱模式-結果02

4、小結

  1. 交換機需要與隊列進行綁定，綁定之後，一個消息可以被多個消費者收到。
  2. 發佈訂閱模式與工作隊列模式的區別：
    • 工作隊列模式不要定義交換機，而發佈/訂閱模式需要定義交換機
    • 工作隊列模式的生產方是面向隊列發送消息（底層使用預設交換機），發佈/訂閱模式的生產方是面向交換機發送消息。
    • 發佈/訂閱模式需要設置隊列和交換機的綁定，工作隊列模式不需要設置，實際上工作隊列模式會將隊列綁定到預設的交換機。

五、路由模式（Routing）

5.1、模式說明

  • 隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由鍵）。
  • 消息的發送方在向Exchange發送消息時，也必須指定消息的RoutingKey。
  • Exchange不再把消息交給每一個綁定的隊列，而是根據消息的RoutingKey來進行判斷，只有隊列的RoutingKey與消息的RoutingKey完全一致，才會接受到消息。

路由模式

圖解：

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

5.2、代碼舉例

1、生產者

public class Producer_Routing {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建交換機
        String exchangeName = "test_direct";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true, false, false, null);
        
        // 6、創建隊列
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、綁定隊列和交換機
        /*
        queueBind(String queue, String exchange, String routingKey)
        參數：
            1、queue：隊列名稱
            2、exchange：交換機名稱
            3、routingKey：路由鍵，綁定規則
                如果交換機的類型為fanout，routingkey設置為""
         */
        // 隊列1綁定 info warning error
        channel.queueBind(queue1Name, exchangeName, "info");
        channel.queueBind(queue1Name, exchangeName, "warning");
        channel.queueBind(queue1Name, exchangeName, "error");
        // 隊列2綁定 error
        channel.queueBind(queue2Name, exchangeName, "error");
        
        // 8、發送消息
        String body = "日誌信息：張三調用了delete方法...出錯誤了...日誌級別：error...";
        
        channel.basicPublish(exchangeName, "error", null, body.getBytes());

        body = "日誌信息：張三調用了findAll方法...日誌級別：info...";

        channel.basicPublish(exchangeName, "info", null, body.getBytes());
        // 9、釋放資源
        channel.close();
        connection.close();
    }
}

2、消費者

消費者1

public class Consumer_Routing1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        String queue1Name = "test_direct_queue1";
        channel.queueDeclare(queue1Name, true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息列印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

消費者2

public class Consumer_Routing2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建隊列 Queue
        String queue2Name = "test_direct_queue2";
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息保存到資料庫.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

3、結果

先啟動生產者，會創建路由模式的交換機和隊列，然後綁定交換機和隊列並指定對應的路由鍵。之後發送消息，通過交換機判斷消息帶的路由鍵發送到兩個綁定的隊列中。

兩個消費者收到消息然後分別執行自己的後續邏輯。

路由模式-結果01

路由模式-結果02

路由模式-結果03

路由模式-結果04

4、小結

Routing模式要求隊列在綁定交換機時要指定RoutingKey，消息會轉發到符合RoutingKey的隊列。

六、通配符模式（Topic）

6.1、模式說明

  • Topic類型與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定RoutingKey的時候使用通配符！
  • RoutingKey一般都是由一個或多個單片語成，多個單詞之間以“.”分割，例如：item.insert。
  • 通配符規則：#匹配一個或多個詞，*匹配不多不少恰好一個詞，例如：item.#能夠匹配item.insert.abc或者item.insert，item.*只能匹配item.insert。

通配符模式

右邊圖解：

  • 紅色Queue：綁定的是usa.#，因此凡是以usa.開頭的RoutingKey都會被匹配到
  • 黃色Queue：綁定的是#.usa，因此凡是以.news結尾的RoutingKey都會被匹配到

6.2、代碼舉例

1、生產者

public class Producer_Topic {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        // 5、創建交換機
        /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        參數：
            1、exchange：交換機名稱
            2、type：交換機類型    
                DIRECT("direct")：定向
                FANOUT("fanout")：扇形（廣播），發生消息到每一個與之綁定隊列。
                TOPIC("topic")：通配符的方式
                HEADERS("headers")：參數匹配
            3、durable：是否持久化
            4、autoDelete：自動刪除
            5、internal：內部使用。一般false
            6、arguments：參數
            
         */
        String exchangeName = "test_topic";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true, false, false, null);
        
        // 6、創建隊列
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、綁定隊列和交換機
        /*
        queueBind(String queue, String exchange, String routingKey)
        參數：
            1、queue：隊列名稱
            2、exchange：交換機名稱
            3、routingKey：路由鍵，綁定規則
                如果交換機的類型為fanout，routingkey設置為""
         */
        // routing key 系統的名稱.日誌的級別
        // 需求：所有error級別的日誌存入資料庫，所有order系統的日誌存入資料庫
        channel.queueBind(queue1Name, exchangeName, "*.*");
        channel.queueBind(queue2Name, exchangeName, "#.error");
        channel.queueBind(queue2Name, exchangeName, "order.*");
        
        // 8、發送消息
        String body = "日誌信息：張三調用了delete方法...出錯誤了...日誌級別：error...";
        
        channel.basicPublish(exchangeName, "good.address.error", null, body.getBytes());

        body = "日誌信息：張三調用了findAll方法...日誌級別：info...";

        channel.basicPublish(exchangeName, "good.info", null, body.getBytes());

        body = "日誌信息：張三調用了update方法...日誌級別：info...";

        channel.basicPublish(exchangeName, "order.info", null, body.getBytes());
        // 9、釋放資源
        channel.close();
        connection.close();
    }
}

2、消費者

消費者1

public class Consumer_Topic1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        String queue1Name = "test_topic_queue1";
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息列印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

消費者2

public class Consumer_Topic2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        // 2、設置參數
        factory.setHost("192.168.3.100");  // ip 預設值 localhost
        factory.setPort(5673); // 埠 預設值 5672
        factory.setVirtualHost("/"); // 虛擬機 預設值 /
        factory.setUsername("guest"); // 用戶名 預設值 guest
        factory.setPassword("guest"); // 密碼 預設值 guest

        // 3、創建連接 Connection
        Connection connection = factory.newConnection();

        // 4、創建頻道 Channel
        Channel channel = connection.createChannel();

        String queue2Name = "test_topic_queue2";
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("將日誌信息保存到資料庫.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、釋放資源 不需要
    }
}

3、結果

先啟動生產者，會創建通配符模式的交換機和隊列，然後綁定交換機和隊列並指定對應的路由鍵。之後發送消息，通過交換機判斷消息帶的路由鍵發送到兩個綁定的隊列中。

兩個消費者收到消息然後分別執行自己的後續邏輯。

通配符模式-結果01

通配符模式-結果02

通配符模式-結果03

通配符模式-結果04

4、小結

Topic主題模式可以實現Pub/Sub發佈與訂閱模式和Routing路由模式的功能，只是Topic在配置RoutingKey

的時候可以使用通配符，顯得更加靈活。

七、工作模式總結

  1. 簡單模式 HelloWorld

    一個生產者、一個消費者，不需要設置交換機（使用預設的交換機）。

  2. 工作模式 Work Queues

    一個生產者、多個消費者（競爭關係），不需要設置交換機（使用預設的交換機）。

  3. 發佈訂閱模式 Publish/Subscribe

    需要設置類型為fanout的交換機，並且交換機和隊列進行綁定，當發送消息到交換機後，交換機會將消息發送到綁定的隊列。

  4. 路由模式 Routing

    需要設置類型為direct的交換機，交換機和隊列進行綁定，並且指定RoutingKey，當發送消息到交換機後，交換機會根據RoutingKey將消息發送到對應的隊列。

  5. 通配符 Topic

    需要設置類型為topic的交換機，交換機和隊列進行綁定，並且指定通配符方式的RoutingKey，當發送消息到交換機後，交換機會根據RoutingKey將消息發送到對應的隊列。

參考


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 爆肝兩萬字，詳解fastdfs分散式文件系統
    1.學習目標 2.簡介 技術論壇：http://bbs.chinaunix.net/forum-240-1.html 資源地址：https://sourceforge.net/projects/fastdfs/ 源碼地址：https://github.com/happyfish100 FastDFS ...
  • maven 多模塊項目的測試覆蓋率分析 - jacoco 聚合分析
    前言 對於大多數 maven 多模塊化工程，可以使用 Jacoco 這款工具，關於 Jacoco 這款工具，ChatGPT 對它的描述是這樣的： JaCoCo（Java Code Coverage）是一個開源的測試覆蓋率工具，它可以用於幫助開發人員衡量其軟體測試的有效性。它支持多種語言，包括 Jav ...
  • Spring簡介-IOC
    1、Spring 1.1、簡介 Spring:春天 >給軟體行業帶來了春天！ 2002，首次推出了Spring框架的雛形：interf21框架！ Spring框架即以interface21框架為基礎，經過重新設計，並不斷豐富其內涵，於2004年3月24日發佈了1.0正式版。 Rod Johnson ...
  • 又發現了一個有趣的 ChatGPT 玩法，分享給大家
    之前給大家寫過如何將 ChatGPT 接入微信和釘釘，沒看過的可以往公眾號前面的文章翻翻，最近又發現了一個有趣的玩法，周末找時間實現了一下，感覺挺不錯的，分享給大家。 背景 事情的起因是阿粉在朋友圈看到了這樣一條信息，敏感信息已經去掉了，意思很明顯就是將 OpenAI 接入到知識星球了，用戶可以通過 ...
  • Celery框架從入門到精通
    Celery介紹、安裝、基本使用 一、Celery服務 什麼是Celery: Celery是一個簡單、靈活且可靠的，處理消息的分散式系統 Celery可以用來做什麼： 非同步任務 定時任務 延遲任務 Celery的運行原理: 可以不依賴任何服務，通過自身命令，啟動服務 celery服務為其他項目服務提 ...
  • jdk17+spring6下打jar包
    問題描述： 利用pyinstaller對python代碼打包後，dist文件夾中會生成一個xxx.exe可執行文件。打包成功，但運行exe時一閃而過（閃退）。捕捉不對到底是打包錯誤呢，還是其他異常？那麼如何解決？ PS：以上現象在windows系統中會出現，在Linux和mac系統中不會出現。 解決 ...
  • 【c語言】整數拆分
    將一個正整數n拆分成若幹個正整數的和（至少兩個數,n<=100）。 輸入格式: 一個正整數n 輸出格式: 若幹行，每行一個等式（數與數之間要求非降序排列）。最後一行給出解的總個數 輸入樣例: 在這裡給出一組輸入。例如： 4 輸出樣例: 4=1+1+1+1 4=1+1+2 4=1+3 4=2+2 4 ...
  • 併發編程那點兒事
    線程理論 線程和進程的區別 進程 進程是操作系統分配資源的最小單位，每個進程都是一個在運行中的程式，在windows中一個運行的xx.exe就是一個進程，他們都擁有自己獨立的一塊記憶體空間，一個進程可以有多個線程 線程 線程是操作系統調度的最小單元，負責當前進程中程式的執行，一個進程可以運行多個線程， ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類