日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當前位置:首頁 > 科技  > 軟件

RabbitMQ工作模式-Routing路由模式

來源: 責編: 時間:2023-11-20 17:12:25 326觀看
導讀Routing路由模式1、模式說明路由模式特點:隊列與交換機的綁定,不能是任意綁定了,而是要指定一個RoutingKey(路由key)。消息的發(fā)送方在 向 Exchange發(fā)送消息時,也必須指定消息的 RoutingKey。Exchange不再把消息交給每一個綁

NQ228資訊網(wǎng)——每日最新資訊28at.com

Routing路由模式

1、模式說明

路由模式特點:NQ228資訊網(wǎng)——每日最新資訊28at.com

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

NQ228資訊網(wǎng)——每日最新資訊28at.com

圖解:NQ228資訊網(wǎng)——每日最新資訊28at.com

  • 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。NQ228資訊網(wǎng)——每日最新資訊28at.com

在寫案例之前,我們首先定義一下需求:NQ228資訊網(wǎng)——每日最新資訊28at.com

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

(1)生產(chǎn)者

NQ228資訊網(wǎng)——每日最新資訊28at.com

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ù)        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<String, Object> 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設置為""         */        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ā)送消息:NQ228資訊網(wǎng)——每日最新資訊28at.com

NQ228資訊網(wǎng)——每日最新資訊28at.com

發(fā)送消息之后,我們來看看聲明好的交換機:NQ228資訊網(wǎng)——每日最新資訊28at.com

NQ228資訊網(wǎng)——每日最新資訊28at.com

NQ228資訊網(wǎng)——每日最新資訊28at.com

(2)消費者1:專門接收 insert 的消息

NQ228資訊網(wǎng)——每日最新資訊28at.com

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ù)        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<String, Object> arguments)        參數(shù):            1. queue:隊列名稱            2. durable:是否持久化,當mq重啟之后,還在            3. exclusive:                * 是否獨占。只能有一個消費者監(jiān)聽這隊列                * 當Connection關閉時,是否刪除隊列            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);        //不需要關閉資源,因為消費者需要持續(xù)監(jiān)聽隊列信息    }}

(3)消費者2:專門接收 update 的消息

NQ228資訊網(wǎng)——每日最新資訊28at.com

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ù)        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<String, Object> arguments)        參數(shù):            1. queue:隊列名稱            2. durable:是否持久化,當mq重啟之后,還在            3. exclusive:                * 是否獨占。只能有一個消費者監(jiān)聽這隊列                * 當Connection關閉時,是否刪除隊列            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);        //不需要關閉資源,因為消費者需要持續(xù)監(jiān)聽隊列信息    }}

3、測試

啟動所有消費者,然后使用生產(chǎn)者發(fā)送消息;在消費者對應的控制臺可以查看到生產(chǎn)者發(fā)送對應routing key對應隊列的消息;到達按照需要接收的效果。NQ228資訊網(wǎng)——每日最新資訊28at.com

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

NQ228資訊網(wǎng)——每日最新資訊28at.com

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

NQ228資訊網(wǎng)——每日最新資訊28at.com

4、小結(jié)

Routing模式要求隊列在綁定交換機時要指定routing key,消息會轉(zhuǎn)發(fā)到符合routing key的隊列。NQ228資訊網(wǎng)——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-31551-0.htmlRabbitMQ工作模式-Routing路由模式

聲明:本網(wǎng)頁內(nèi)容旨在傳播知識,若有侵權等問題請及時與本網(wǎng)聯(lián)系,我們將在第一時間刪除處理。郵件:2376512515@qq.com

上一篇: 快速了解 CSS @starting-style 規(guī)則

下一篇: 通過實例理解Web應用跨域問題

標簽:
  • 熱門焦點
  • 俄羅斯:將審查iPhone等外國公司設備 保數(shù)據(jù)安全

    iPhone和特斯拉都屬于在各自領域領頭羊的品牌,推出的產(chǎn)品也也都是數(shù)一數(shù)二的,但對于一些國家而言,它們的產(chǎn)品可靠性和安全性還是在限制范圍內(nèi)。近日,俄羅斯聯(lián)邦通信、信息技術
  • 影音體驗是真的強 簡單聊聊iQOO Pad

    大公司的好處就是產(chǎn)品線豐富,非常細分化的東西也能給你做出來,例如早先我們看到了新的vivo Pad2,之后我們又在iQOO Neo8 Pro的發(fā)布會上看到了iQOO的首款平板產(chǎn)品iQOO Pad。雖
  • 6月安卓手機性能榜:vivo/iQOO霸占旗艦排行榜前三

    2023年上半年已經(jīng)正式過去了,我們也迎來了安兔兔V10版本,在新的驍龍8Gen3和天璣9300發(fā)布之前,性能榜的榜單大體會以驍龍8Gen2和天璣9200+為主,至于那顆3.36GHz的驍龍8Gen2領先
  • 一文看懂為蘋果Vision Pro開發(fā)應用程序

    譯者 | 布加迪審校 | 重樓蘋果的Vision Pro是一款混合現(xiàn)實(MR)頭戴設備。Vision Pro結(jié)合了虛擬現(xiàn)實(VR)和增強現(xiàn)實(AR)的沉浸感。其高分辨率顯示屏、先進的傳感器和強大的處理能力
  • JavaScript學習 -AES加密算法

    引言在當今數(shù)字化時代,前端應用程序扮演著重要角色,用戶的敏感數(shù)據(jù)經(jīng)常在前端進行加密和解密操作。然而,這樣的操作在網(wǎng)絡傳輸和存儲中可能會受到惡意攻擊的威脅。為了確保數(shù)據(jù)
  • iQOO 11S新品發(fā)布會

    iQOO將在7月4日19:00舉行新品發(fā)布會,推出杭州亞運會電競賽事官方用機iQOO 11S。
  • OPPO K11搭載長壽版100W超級閃充:26分鐘充滿100%

    據(jù)此前官方宣布,OPPO將于7月25日也就是今天下午14:30舉辦新品發(fā)布會,屆時全新的OPPO K11將正式與大家見面,將主打旗艦影像,和同檔位競品相比,其最大的賣
  • 世界人工智能大會國際日開幕式活動在世博展覽館開啟

    30日上午,世界人工智能大會國際日開幕式活動在世博展覽館開啟,聚集國際城市代表、重量級院士專家、國際創(chuàng)新企業(yè)代表,共同打造人工智能交流平臺。上海市副市
  • 親歷馬斯克血洗Twitter,硅谷的苦日子在后頭

    文/劉哲銘  編輯/李薇  馬斯克再次揮下裁員大刀?! ∶绹鴷r間11月14日,Twitter約4400名外包員工遭解雇,此次被解雇的員工的主要工作為內(nèi)容審核等。此前,T
Top 主站蜘蛛池模板: 兴城市| 白城市| 诏安县| 长丰县| 台东市| 武鸣县| 宽城| 林周县| 策勒县| 永宁县| 额敏县| 丹凤县| 洞口县| 漳浦县| 原平市| 榆树市| 信宜市| 柳河县| 乃东县| 黑河市| 榆中县| 保定市| 普兰县| 巴彦淖尔市| 沂南县| 布拖县| 东辽县| 琼海市| 和政县| 巴彦淖尔市| 安图县| 黄冈市| 云阳县| 城固县| 延边| 栾川县| 朝阳市| 密山市| 城步| 富源县| 松滋市|