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 6 Entity Framework Core Db First（1） 創建模型
    .Net8.0 Blazor Hybird 桌面端 (WPF/Winform) 實測可以完整運行在 win7sp1/win10/win11. 如果用其他工具打包,還可以運行在mac/linux下, 傳送門BlazorHybrid 發佈為無依賴包方式 安裝 WebView2Runtime 1.57 M ...
  • 資料庫升級PostgreSql+Garnet
    目錄前言PostgreSql安裝測試額外Nuget安裝Person.cs模擬運行Navicate連postgresql解決方案Garnet為什麼要選擇Garnet而不是RedisRedis不再開源Windows版的Redis是由微軟維護的Windows Redis版本老舊，後續可能不再更新Garne ...
  • 如何基於surging架設流媒體視頻推流（視頻講解）
    C#TMS系統代碼-聯表報表學習 領導被裁了之後很快就有人上任了，幾乎是無縫銜接，很難讓我不想到這早就決定好了。我的職責沒有任何變化。感受下來這個系統封裝程度很高，我只要會調用方法就行。這個系統交付之後不會有太多問題，更多應該是做小需求，有大的開發任務應該也是第二期的事，嗯？怎麼感覺我變成運維了？而 ...
  • EAV模型（實體-屬性-值）的設計和低代碼的處理方案（2）--數據的查詢處理
    我在隨筆《EAV模型（實體-屬性-值）的設計和低代碼的處理方案（1）》中介紹了一些基本的EAV模型設計知識和基於Winform場景下低代碼（或者說無代碼）的一些實現思路，在本篇隨筆中，我們來分析一下這種針對通用業務，且只需定義就能構建業務模塊存儲和界面的解決方案，其中的數據查詢處理的操作。 ...
  • C# 程式集、模塊和類型概念及關係
    對某個遠程伺服器啟用和設置NTP服務(Windows系統) 打開註冊表 HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\W32Time\TimeProviders\NtpServer 將 Enabled 的值設置為 1，這將啟用NTP伺服器功 ...
  • Django信號與擴展：深入理解與實踐
    title: Django信號與擴展：深入理解與實踐 date: 2024/5/15 22:40:52 updated: 2024/5/15 22:40:52 categories: 後端開發 tags: Django 信號 松耦合 觀察者 擴展 安全 性能 第一部分：Django信號基礎 Djan ...
  • Django3.2安裝源碼xadmin2步驟以及遇到的問題
    使用xadmin2遇到的問題&解決 環境配置： 使用的模塊版本： 關聯的包 Django 3.2.15 mysqlclient 2.2.4 xadmin 2.0.1 django-crispy-forms >= 1.6.0 django-import-export >= 0.5.1 django-r ...
  • 原創->CommonsCollections1-DefaultMap鏈
    今天我打算整點兒不一樣的內容，通過之前學習的TransformerMap和LazyMap鏈，想搞點不一樣的，所以我關註了另外一條鏈DefaultedMap鏈，主要調用鏈為： 調用鏈詳細描述： ObjectInputStream.readObject() DefaultedMap.readObject ...
  • 05-應用級開發者 AI 時代破局點
    後端應用級開發者該如何擁抱 AI GC？就是在這樣的一個大的浪潮下，我們的傳統的應用級開發者。我們該如何選擇職業或者是如何去快速轉型，跟上這樣的一個行業的一個浪潮? 0 AI金字塔模型 越往上它的整個難度就是職業機會也好，或者說是整個的這個運作也好,它的難度會越大,然後越往下機會就會越多,所以這是一 ...
  • Java面試題：讓依賴註入變得簡單，面對@Autowired和@Resource，該如何選擇？
    @Autowired是Spring框架提供的註解，@Resource是Java EE 5規範提供的註解。 @Autowired預設按照類型自動裝配，而@Resource預設按照名稱自動裝配。 @Autowired支持@Qualifier註解來指定裝配哪一個具有相同類型的bean，而@Resourc... ...
所有分類