日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
RabbitMQ工作模式-Routing路由模式

Routing路由模式

1、模式說明

路由模式特點：

富源網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)！從網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、成都響應(yīng)式網(wǎng)站建設(shè)等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)自2013年創(chuàng)立以來到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進行。專注于網(wǎng)站建設(shè)就選創(chuàng)新互聯(lián)。

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

圖解：

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

2。案例

在編碼上與 Publish/Subscribe發(fā)布與訂閱模式 的區(qū)別是交換機的類型為：Direct，還有隊列綁定交換機的時候需要指定routing key。

在寫案例之前，我們首先定義一下需求：

  • 生產(chǎn)者：發(fā)送兩條消息，一條消息的用于插入數(shù)據(jù)，另一條消息用于更新數(shù)據(jù)。
  • 消費者1：接收插入數(shù)據(jù)的消息，進行數(shù)據(jù)插入。
  • 消費者2：接收更新數(shù)據(jù)的消息，進行數(shù)據(jù)更新。

（1）生產(chǎn)者

package com.lijw.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @author Aron.li
 * @date 2022/3/3 8:16
 */
public class Producer_Routing {
    //交換機名稱
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //隊列名稱
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //隊列名稱
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設(shè)置參數(shù)
        factory.setHost("127.0.0.1"); // ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/test"); //虛擬機 默認值 /
        factory.setUsername("libai"); // 用戶名 默認 guest
        factory.setPassword("libai"); //密碼 默認值 guest
        //3. 創(chuàng)建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創(chuàng)建Channel
        Channel channel = connection.createChannel();
        //5. 創(chuàng)建交換機
        /*
           exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map arguments)
           參數(shù)：
            1. exchange:交換機名稱
            2. type:交換機類型
                DIRECT("direct")：定向
                FANOUT("fanout")：扇形（廣播），發(fā)送消息到每一個與之綁定隊列。
                TOPIC("topic") 通配符的方式
                HEADERS("headers") 參數(shù)匹配
            3. durable:是否持久化
            4. autoDelete:自動刪除
            5. internal：內(nèi)部使用。 一般false
            6. arguments：參數(shù)
        */
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT, true, false, false, null);
        // 6.聲明（創(chuàng)建）隊列
        /**
         * 參數(shù)1：隊列名稱
         * 參數(shù)2：是否定義持久化隊列
         * 參數(shù)3：是否獨占本次連接
         * 參數(shù)4：是否在不使用的時候自動刪除隊列
         * 參數(shù)5：隊列其它參數(shù)
         */
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
        // 7. 綁定隊列和交換機
        /*
            queueBind(String queue, String exchange, String routingKey)
            參數(shù)：
                1. queue：隊列名稱
                2. exchange：交換機名稱
                3. routingKey：路由鍵，綁定規(guī)則
                    如果交換機的類型為fanout ，routingKey設(shè)置為""
         */
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
        channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
        //8. 發(fā)送消息至交換機，由交換機分發(fā)消息
        // 發(fā)送信息
        String message = "新增了商品。路由模式；routing key 為 insert " ;
        /**
         * 參數(shù)1：交換機名稱，如果沒有指定則使用默認Default Exchage
         * 參數(shù)2：路由key,簡單模式可以傳遞隊列名稱
         * 參數(shù)3：消息其它屬性
         * 參數(shù)4：消息內(nèi)容
         */
        channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
        System.out.println("已發(fā)送消息：" + message);
        // 發(fā)送信息
        message = "修改了商品。路由模式；routing key 為 update" ;
        /**
         * 參數(shù)1：交換機名稱，如果沒有指定則使用默認Default Exchage
         * 參數(shù)2：路由key,簡單模式可以傳遞隊列名稱
         * 參數(shù)3：消息其它屬性
         * 參數(shù)4：消息內(nèi)容
         */
        channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
        System.out.println("已發(fā)送消息：" + message);
        //9. 釋放資源
        channel.close();
        connection.close();
    }
}

執(zhí)行發(fā)送消息：

發(fā)送消息之后，我們來看看聲明好的交換機：

（2）消費者1：專門接收 insert 的消息

package com.lijw.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author Aron.li
 * @date 2022/3/2 16:16
 */
public class Consumer_Routing1 {

    //隊列名稱
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";

    public static void main(String[] args) throws IOException, TimeoutException {
        //1.創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設(shè)置參數(shù)
        factory.setHost("127.0.0.1"); // ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/test"); //虛擬機 默認值 /
        factory.setUsername("libai"); // 用戶名 默認 guest
        factory.setPassword("libai"); //密碼 默認值 guest
        //3. 創(chuàng)建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創(chuàng)建Channel
        Channel channel = connection.createChannel();
        //5. 創(chuàng)建隊列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        參數(shù)：
            1. queue：隊列名稱
            2. durable:是否持久化，當mq重啟之后，還在
            3. exclusive：
                * 是否獨占。只能有一個消費者監(jiān)聽這隊列
                * 當Connection關(guān)閉時，是否刪除隊列
            4. autoDelete:是否自動刪除。當沒有Consumer時，自動刪除掉
            5. arguments：參數(shù)。

         */
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);

        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        參數(shù)：
            1. queue：隊列名稱
            2. autoAck：是否自動確認
            3. callback：回調(diào)對象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回調(diào)方法，當收到消息后，會自動執(zhí)行該方法
                1. consumerTag：標識
                2. envelope：獲取一些信息，交換機，路由key...
                3. properties:配置信息
                4. body：數(shù)據(jù)
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收隊列的數(shù)據(jù) body: " + new String(body));
            }
        };
        channel.basicConsume(DIRECT_QUEUE_INSERT,true,consumer);

        //不需要關(guān)閉資源，因為消費者需要持續(xù)監(jiān)聽隊列信息
    }
}

（3）消費者2：專門接收 update 的消息

package com.lijw.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @author Aron.li
 * @date 2022/3/2 16:16
 */
public class Consumer_Routing2 {
    //隊列名稱
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設(shè)置參數(shù)
        factory.setHost("127.0.0.1"); // ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/test"); //虛擬機 默認值 /
        factory.setUsername("libai"); // 用戶名 默認 guest
        factory.setPassword("libai"); //密碼 默認值 guest
        //3. 創(chuàng)建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創(chuàng)建Channel
        Channel channel = connection.createChannel();
        //5. 創(chuàng)建隊列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        參數(shù)：
            1. queue：隊列名稱
            2. durable:是否持久化，當mq重啟之后，還在
            3. exclusive：
                * 是否獨占。只能有一個消費者監(jiān)聽這隊列
                * 當Connection關(guān)閉時，是否刪除隊列
            4. autoDelete:是否自動刪除。當沒有Consumer時，自動刪除掉
            5. arguments：參數(shù)。
         */
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        參數(shù)：
            1. queue：隊列名稱
            2. autoAck：是否自動確認
            3. callback：回調(diào)對象
         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回調(diào)方法，當收到消息后，會自動執(zhí)行該方法
                1. consumerTag：標識
                2. envelope：獲取一些信息，交換機，路由key...
                3. properties:配置信息
                4. body：數(shù)據(jù)
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收隊列的數(shù)據(jù) body: " + new String(body));
            }
        };
        channel.basicConsume(DIRECT_QUEUE_UPDATE,true,consumer);
        //不需要關(guān)閉資源，因為消費者需要持續(xù)監(jiān)聽隊列信息
    }
}

3、測試

啟動所有消費者，然后使用生產(chǎn)者發(fā)送消息；在消費者對應(yīng)的控制臺可以查看到生產(chǎn)者發(fā)送對應(yīng)routing key對應(yīng)隊列的消息；到達按照需要接收的效果。

  • 消費者1 收到了 insert 的消息

  • 消費者2 收到了 update 的消息

4、小結(jié)

Routing模式要求隊列在綁定交換機時要指定routing key，消息會轉(zhuǎn)發(fā)到符合routing key的隊列。


網(wǎng)站題目：RabbitMQ工作模式-Routing路由模式
分享鏈接：http://www.dlmjj.cn/article/cdhehoi.html