2. 程式人生 > 實用技巧 >RabbitMQ工作模式-Routing路由模式

RabbitMQ工作模式-Routing路由模式

阿新 • • 發佈：2020-10-21

RabbitMQ工作模式-Routing路由模式

路由模式： 1、每個消費者監聽自己的佇列，並且設定routingkey。 2、生產者將訊息發給交換機，由交換機根據routingkey來轉發訊息到指定的佇列，消費者佇列中的routingkey（佇列和交換機繫結時指定，可指定多個）只有匹配上才可以消費。

  • 生產者
package com.xuecheng.test.rabbitmq;

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 Administrator
 * @version 1.0
 * @create 2018-06-17 19:23
 **/
public class Producer03_routing {
    //佇列名稱
    private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    private static final String EXCHANGE_ROUTING_INFORM="
 
exchange_routing_inform";
    private static final String ROUTINGKEY_EMAIL="inform_email";
    private static final String ROUTINGKEY_SMS="inform_sms";
    public static void main(String[] args) {
        //通過連線工廠建立新的連線和mq建立連線
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(
 
"127.0.0.1");
        connectionFactory.setPort(5672);//埠
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //設定虛擬機器，一個mq服務可以設定多個虛擬機器，每個虛擬機器就相當於一個獨立的mq
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        try {
            //建立新連線
            connection = connectionFactory.newConnection();
            //建立會話通道,生產者和mq服務所有通訊都在channel通道中完成
            channel = connection.createChannel();
            //宣告佇列，如果佇列在mq 中沒有則要建立
            //引數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            /**
             * 引數明細
             * 1、queue 佇列名稱
             * 2、durable 是否持久化，如果持久化，mq重啟後佇列還在
             * 3、exclusive 是否獨佔連線，佇列只允許在該連線中訪問，如果connection連線關閉佇列則自動刪除,如果將此引數設定true可用於臨時佇列的建立
             * 4、autoDelete 自動刪除，佇列不再使用時是否自動刪除此佇列，如果將此引數和exclusive引數設定為true就可以實現臨時佇列（佇列不用了就自動刪除）
             * 5、arguments 引數，可以設定一個佇列的擴充套件引數，比如：可設定存活時間
             */
            channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
            channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
            //宣告一個交換機
            //引數：String exchange, String type
            /**
             * 引數明細：
             * 1、交換機的名稱
             * 2、交換機的型別
             * fanout：對應的rabbitmq的工作模式是 publish/subscribe
             * direct：對應的Routing    工作模式
             * topic：對應的Topics工作模式
             * headers： 對應的headers工作模式
             */
            channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
            //進行交換機和佇列繫結
            //引數：String queue, String exchange, String routingKey
            /**
             * 引數明細：
             * 1、queue 佇列名稱
             * 2、exchange 交換機名稱
             * 3、routingKey 路由key，作用是交換機根據路由key的值將訊息轉發到指定的佇列中，在釋出訂閱模式中調協為空字串
             */
            channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);
            channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,"inform");
            channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);
            channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,"inform");
            //傳送訊息
            //引數：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * 引數明細：
             * 1、exchange，交換機，如果不指定將使用mq的預設交換機（設定為""）
             * 2、routingKey，路由key，交換機根據路由key來將訊息轉發到指定的佇列，如果使用預設交換機，routingKey設定為佇列的名稱
             * 3、props，訊息的屬性
             * 4、body，訊息內容
             */
           /* for(int i=0;i<5;i++){
                //傳送訊息的時候指定routingKey
                String message = "send email inform message to user";
                channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL,null,message.getBytes());
                System.out.println("send to mq "+message);
            }
            for(int i=0;i<5;i++){
                //傳送訊息的時候指定routingKey
                String message = "send sms inform message to user";
                channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes());
                System.out.println("send to mq "+message);
            }*/
            for(int i=0;i<5;i++){
                //傳送訊息的時候指定routingKey
                String message = "send inform message to user";
                channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes());
                System.out.println("send to mq "+message);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //關閉連線
            //先關閉通道
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }


    }
}

  • 消費者-email
package com.xuecheng.test.rabbitmq;

import com.rabbitmq.client.*;

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

/**
 * @author Administrator
 * @version 1.0
 * @create 2018-06-17 18:22
 **/
public class Consumer03_routing_email {
    //佇列名稱
    private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";
    private static final String ROUTINGKEY_EMAIL="inform_email";

    public static void main(String[] args) throws IOException, TimeoutException {
        //通過連線工廠建立新的連線和mq建立連線
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);//埠
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //設定虛擬機器，一個mq服務可以設定多個虛擬機器，每個虛擬機器就相當於一個獨立的mq
        connectionFactory.setVirtualHost("/");

        //建立新連線
        Connection connection = connectionFactory.newConnection();
        //建立會話通道,生產者和mq服務所有通訊都在channel通道中完成
        Channel channel = connection.createChannel();

        /**
         * 引數明細
         * 1、queue 佇列名稱
         * 2、durable 是否持久化，如果持久化，mq重啟後佇列還在
         * 3、exclusive 是否獨佔連線，佇列只允許在該連線中訪問，如果connection連線關閉佇列則自動刪除,如果將此引數設定true可用於臨時佇列的建立
         * 4、autoDelete 自動刪除，佇列不再使用時是否自動刪除此佇列，如果將此引數和exclusive引數設定為true就可以實現臨時佇列（佇列不用了就自動刪除）
         * 5、arguments 引數，可以設定一個佇列的擴充套件引數，比如：可設定存活時間
         */
        channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
        //宣告一個交換機
        //引數：String exchange, String type
        /**
         * 引數明細：
         * 1、交換機的名稱
         * 2、交換機的型別
         * fanout：對應的rabbitmq的工作模式是 publish/subscribe
         * direct：對應的Routing    工作模式
         * topic：對應的Topics工作模式
         * headers： 對應的headers工作模式
         */
        channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
        //進行交換機和佇列繫結
        //引數：String queue, String exchange, String routingKey
        /**
         * 引數明細：
         * 1、queue 佇列名稱
         * 2、exchange 交換機名稱
         * 3、routingKey 路由key，作用是交換機根據路由key的值將訊息轉發到指定的佇列中，在釋出訂閱模式中調協為空字串
         */
        channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);

        //實現消費方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){

            /**
             * 當接收到訊息後此方法將被呼叫
             * @param consumerTag  消費者標籤，用來標識消費者的，在監聽佇列時設定channel.basicConsume
             * @param envelope 信封，通過envelope
             * @param properties 訊息屬性
             * @param body 訊息內容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交換機
                String exchange = envelope.getExchange();
                //訊息id，mq在channel中用來標識訊息的id，可用於確認訊息已接收
                long deliveryTag = envelope.getDeliveryTag();
                //訊息內容
                String message= new String(body,"utf-8");
                System.out.println("receive message:"+message);
            }
        };

        //監聽佇列
        //引數：String queue, boolean autoAck, Consumer callback
        /**
         * 引數明細：
         * 1、queue 佇列名稱
         * 2、autoAck 自動回覆，當消費者接收到訊息後要告訴mq訊息已接收，如果將此引數設定為tru表示會自動回覆mq，如果設定為false要通過程式設計實現回覆
         * 3、callback，消費方法，當消費者接收到訊息要執行的方法
         */
        channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);

    }
}

  • 消費者sms：
package com.xuecheng.test.rabbitmq;

import com.rabbitmq.client.*;

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

/**
 * @author Administrator
 * @version 1.0
 * @create 2018-06-17 18:22
 **/
public class Consumer03_routing_sms {
    //佇列名稱
    private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";
    private static final String ROUTINGKEY_SMS="inform_sms";

    public static void main(String[] args) throws IOException, TimeoutException {
        //通過連線工廠建立新的連線和mq建立連線
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);//埠
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //設定虛擬機器，一個mq服務可以設定多個虛擬機器，每個虛擬機器就相當於一個獨立的mq
        connectionFactory.setVirtualHost("/");

        //建立新連線
        Connection connection = connectionFactory.newConnection();
        //建立會話通道,生產者和mq服務所有通訊都在channel通道中完成
        Channel channel = connection.createChannel();

        /**
         * 引數明細
         * 1、queue 佇列名稱
         * 2、durable 是否持久化，如果持久化，mq重啟後佇列還在
         * 3、exclusive 是否獨佔連線，佇列只允許在該連線中訪問，如果connection連線關閉佇列則自動刪除,如果將此引數設定true可用於臨時佇列的建立
         * 4、autoDelete 自動刪除，佇列不再使用時是否自動刪除此佇列，如果將此引數和exclusive引數設定為true就可以實現臨時佇列（佇列不用了就自動刪除）
         * 5、arguments 引數，可以設定一個佇列的擴充套件引數，比如：可設定存活時間
         */
        channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
        //宣告一個交換機
        //引數：String exchange, String type
        /**
         * 引數明細：
         * 1、交換機的名稱
         * 2、交換機的型別
         * fanout：對應的rabbitmq的工作模式是 publish/subscribe
         * direct：對應的Routing    工作模式
         * topic：對應的Topics工作模式
         * headers： 對應的headers工作模式
         */
        channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
        //進行交換機和佇列繫結
        //引數：String queue, String exchange, String routingKey
        /**
         * 引數明細：
         * 1、queue 佇列名稱
         * 2、exchange 交換機名稱
         * 3、routingKey 路由key，作用是交換機根據路由key的值將訊息轉發到指定的佇列中，在釋出訂閱模式中調協為空字串
         */
        channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);

        //實現消費方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){

            /**
             * 當接收到訊息後此方法將被呼叫
             * @param consumerTag  消費者標籤，用來標識消費者的，在監聽佇列時設定channel.basicConsume
             * @param envelope 信封，通過envelope
             * @param properties 訊息屬性
             * @param body 訊息內容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交換機
                String exchange = envelope.getExchange();
                //訊息id，mq在channel中用來標識訊息的id，可用於確認訊息已接收
                long deliveryTag = envelope.getDeliveryTag();
                //訊息內容
                String message= new String(body,"utf-8");
                System.out.println("receive message:"+message);
            }
        };

        //監聽佇列
        //引數：String queue, boolean autoAck, Consumer callback
        /**
         * 引數明細：
         * 1、queue 佇列名稱
         * 2、autoAck 自動回覆，當消費者接收到訊息後要告訴mq訊息已接收，如果將此引數設定為tru表示會自動回覆mq，如果設定為false要通過程式設計實現回覆
         * 3、callback，消費方法，當消費者接收到訊息要執行的方法
         */
        channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer);

    }
}

經過測試 路由模式具備釋出訂閱模式，即當兩個佇列在繫結交換機的時候配置了相同的routingKey，兩者都會接受相同的訊息

相關推薦

RabbitMQ工作模式-Routing路由模式

RabbitMQ工作模式-Routing路由模式 路由模式： 1、每個消費者監聽自己的佇列，並且設定routingkey。

《RabbitMQ系列教程-第四章-04-RabbitMQ工作模式之Routing路由模式》

技術標籤：# 《RabbitMQ系列教程》佇列交換機rabbitmqjavaqueue RabbitMQ工作模式之Routing路由模式

RabbitMQ (六) 訂閱者模式之路由模式 ( direct )

路由模式下,生產者傳送訊息時需要指定一個路由鍵(routingKey),交換機只會把訊息轉發給包含該路由鍵的佇列

Vue 路由 基本操作 路由模式 hash路由模式的實現原理

1-路由 1.1-路由的概念 頁面訪問路徑和元件之間的對應關係 1.2-路由的安裝 npm i vue-router -S

rabbitmq工作模式---路由模式

模式說明: 佇列與交換機的繫結，不能是任意綁定了，而是要指定一個RoutingKey（路由key）

rabbitmq工作模式（一）釋出訂閱模式

專案碼雲地址：https://gitee.com/menbbo/springboot-rabbitmq.git RabbitMQ的工作模式包括了：工作佇列模式、釋出訂閱模式、路由模式、TOPIC（萬用字元模式）。

rabbitmq工作模式（三）Topics萬用字元模式

RabbitMQ的工作模式包括了：簡單模式、工作佇列模式、釋出訂閱模式、路由模式、TOPICS（萬用字元模式）以及RPC。本文主要介紹topics模式。

RabbitMQ——工作佇列模式

1.模式說明 應用場景：對於 任務過重或任務較多情況使用工作佇列可以提高任務處理的速度 只有3個角色：P：生產者，也就是要傳送訊息的程式C：消費者：訊息的接受者，會一直等待訊息到來。queue：訊息佇列，圖中紅色

實戰阿里雲-路由模式工作並埠轉發（埠對映）

1、前言 筆者阿里雲後端增加一臺windows的主機，為了安全與降低成本，後端主機沒有公網網絡卡，於是筆者實戰阿里雲的埠轉發（解決前端主機以路由模式工作並做埠對映問題）。

RabbitMQ工作模式-釋出訂閱

RabbitMQ工作模式-釋出訂閱 釋出訂閱模式： 1、每個消費者監聽自己的佇列。 2、生產者將訊息發給broker，由交換機將訊息轉發到繫結此交換機的每個佇列，每個繫結交換機的佇列都將接收

RabbitMQ工作模式-Topic模式

RabbitMQ工作模式-Topic模式 1、每個消費者監聽自己的佇列，並且設定帶統配符的routingkey。

【夏目鬼鬼分享】RabbitMQ路由模式

技術標籤：rabbitmq 路由模式 路由模式是可以根據路由鍵選擇性給多個消費者傳送訊息的模式，它包含一個生產者、兩個消費者、兩個佇列和一個交換機。兩個消費者同時繫結到不同的佇列上去，兩個佇列通過路由鍵繫結

《RabbitMQ系列教程-第四章-05-RabbitMQ工作模式之Topics主題模式》

技術標籤：# 《RabbitMQ系列教程》佇列交換機rabbitmqjavaqueue RabbitMQ工作模式之Topics主題模式

《RabbitMQ系列教程-第四章-03-RabbitMQ工作模式之Pub/Sub釋出訂閱模式》

技術標籤：# 《RabbitMQ系列教程》交換機佇列rabbitmqjavaqueue RabbitMQ工作模式之Pub/Sub釋出訂閱模式

《RabbitMQ系列教程-第四章-01-RabbitMQ工作模式之Simple模式》

技術標籤：# 《RabbitMQ系列教程》佇列rabbitmqjava訊息佇列交換機 RabbitMQ工作模式之Simple模式

訊息中介軟體-RabbitMQ(工作模式)

技術標籤：訊息中介軟體rabbitmq交換機java Pub/Sub（訂閱模式）P：生產者，也就是要傳送訊息的程式，但是不再發送到佇列中，而是發給X（交換機） C：消費者，訊息的接收者，會一直等待訊息到來 Queue：訊息佇列

springboot整合rabbitmq（direct路由模式）

生產者 在fanout基礎上更改；（只給routeKey是email和duanxin的發） 在OrderService新增如下方法

RabbitMQ - 常用訊息佇列之路由模式【Direct Exchange】

路由模式介紹 　　路由模式下,在釋出訊息時指定不同的routeKey,交換機會根據不同的routeKey分發訊息到不同的佇列中

RabbitMQ之路由模式

RabbitMQ之路由模式 概念 簡單來說就是控制消費者拿到特定條件的訊息 比如一個情景:生產者生產日誌訊息 然後低級別的日誌交給一號消費者處理 嚴重的交給二號消費者處理

工作流資料模式-基於資料的路由模式7種

版權宣告：工作流模式版權歸 Workflow Patterns 組 織 ( http://www.workflowpatterns.com ) 所 有 。 經 Workflow Patterns授權，中文簡體版由辛鵬和榮浩翻譯。未經譯者書面許可，不得將該中文簡體版用於商業目的