RabbitMQ實戰教程(五)-路由模式

阿新 • • 發佈：2018-12-10

1. 路由模式

跟訂閱模式類似,只不過在訂閱模式的基礎上加上路由,訂閱模式是分發到所有繫結到該交換機的佇列,路由模式只分發到繫結在該交換機上面指定的路由鍵佇列.

2. Direct 直接交換機

直連交換機(Direct Exchange)是一種帶路由功能的交換機,它將訊息中的Routing Key與該交換機關聯的所有Binding中的Routing Key進行比較,如果完全相等將訊息傳送到Binding對應的佇列中 .

適用場景 : 根據任務的優先順序把訊息傳送到對應的佇列中,分配更多資源處理優先順序高的佇列.

生產者宣告一個direct型別的交換機,然後傳送訊息到這個交換機指定路由鍵. 消費者指定消費交換機的路由鍵,即可以接到到訊息,其他消費者接收不到 .

在Fanout交換機中

生產者 :

// 第二個引數就是路由鍵
channel.basicPublish(EXCHANE_NAME,"",null,message.getBytes());

消費者 :

// 第三個引數就是路由鍵
channel.queueBind(QUEUE,EXCAHNGE_NAME,"");

3. 程式碼演示

同樣的,只是交換機型別改為driect,加了個路由鍵而已 .

這裡演示3個,即表示一個日誌事件,根據日誌型別進行處理

3.1 工具類

package com.makesailing.neo.utils;

import com.rabbitmq.client. 
Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # RabbitMQ連線工具類
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 14:38
 */
public class ConnectionUtils {

  public static final String host = "127.0.0.1";

  public 
 static final Integer port = 5672;

  public static Connection getConnection() throws IOException, TimeoutException {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost(host);
    connectionFactory.setPort(port);
    // 如果有 使用者名稱 密碼 vhost 配置即可
    connectionFactory.setUsername("jamie");
    connectionFactory.setPassword("123456");
    connectionFactory.setVirtualHost("/simple");
    return connectionFactory.newConnection();
  }
}

3.2 日誌生產者

package com.makesailing.neo.provider;

import com.makesailing.neo.utils.ConnectionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # 日誌訊息 提供者
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 22:41
 */
public class LogSend {

  private static final String EXCHANGE_NAME = "test_exchange_direct";

  private static final String INTO_ROUTING_NAME= "info";
  private static final String WARN_ROUTING_NAME= "warn";
  private static final String ERROR_ROUTING_NAME= "error";


  public static void main(String[] args) throws IOException, TimeoutException {
    // 建立連線
    Connection connection = ConnectionUtils.getConnection();
    // 獲取通道
    Channel channel = connection.createChannel();
    // 宣告一個direct 路由交換機
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

    // 傳送info路由鍵訊息
    String infoMessage = "Hello RabbitMQ Info Log";
    channel.basicPublish(EXCHANGE_NAME, INTO_ROUTING_NAME, null, infoMessage.getBytes());
    System.out.println("  LogSend routing info message : '" + infoMessage + "'");
    // 傳送warn路由鍵訊息
    String warnMessage = "Hello RabbitMQ Warn Log";
    channel.basicPublish(EXCHANGE_NAME, WARN_ROUTING_NAME, null, warnMessage.getBytes());
    System.out.println("  LogSend routing warn message : '" + warnMessage + "'");
    // 傳送info路由鍵訊息
    String errorMessage = "Hello RabbitMQ Error Log";
    channel.basicPublish(EXCHANGE_NAME, ERROR_ROUTING_NAME, null, errorMessage.getBytes());
    System.out.println("  LogSend routing error message : '" + errorMessage + "'");

    channel.close();
    connection.close();
  }
}

3.3 error日誌消費者

package com.makesailing.neo.consumer;

import com.makesailing.neo.utils.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # 錯誤日誌 消費者
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 23:09
 */
public class ErrorReceive {
  private static final String EXCHANGE_NAME = "test_exchange_direct";
  // info日誌佇列
  private static final String QUEUE_NAME = "test_queue_routing_error";

  private static final String ERROR_ROUTING_NAME= "error";

  public static void main(String[] args) throws IOException, TimeoutException {
    // 獲取連線
    Connection connection = ConnectionUtils.getConnection();

    // 開啟通道
    Channel channel = connection.createChannel();

    // 申明要消費的佇列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    // 繫結佇列到交換機
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ERROR_ROUTING_NAME);

    // 這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說，在接收到該Consumer的ack前，他它不會將新的Message分發給它。
    channel.basicQos(1);

    // 建立一個回撥的消費者處理類
    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 接收到的訊息
        String message = new String(body);
        System.out.println(" ErrorReceive '" + message + "' , 任務處理中");

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        } finally {
          System.out.println(" ErrorReceive done ");
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

    // 消費訊息
    channel.basicConsume(QUEUE_NAME, false, consumer);

  }
}

3.4 info日誌消費者

package com.makesailing.neo.consumer;

import com.makesailing.neo.utils.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # info 日誌消費者
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 23:02
 */
public class InfoReceive {

  private static final String EXCHANGE_NAME = "test_exchange_direct";
  // info日誌佇列
  private static final String QUEUE_NAME = "test_queue_routing_info";

  private static final String INTO_ROUTING_NAME= "info";

  public static void main(String[] args) throws IOException, TimeoutException {
    // 獲取連線
    Connection connection = ConnectionUtils.getConnection();

    // 開啟通道
    Channel channel = connection.createChannel();

    // 申明要消費的佇列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    // 繫結佇列到交換機
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, INTO_ROUTING_NAME);

    // 這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說，在接收到該Consumer的ack前，他它不會將新的Message分發給它。
    channel.basicQos(1);

    // 建立一個回撥的消費者處理類
    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 接收到的訊息
        String message = new String(body);
        System.out.println(" InfoReceive '" + message + "' , 任務處理中");

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        } finally {
          System.out.println(" InfoReceive done ");
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

    // 消費訊息
    channel.basicConsume(QUEUE_NAME, false, consumer);

  }
}

3.5 warn日誌消費者

package com.makesailing.neo.consumer;

import com.makesailing.neo.utils.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # 警告日誌 消費者
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 23:09
 */
public class WarnReceive {
  private static final String EXCHANGE_NAME = "test_exchange_direct";
  // info日誌佇列
  private static final String QUEUE_NAME = "test_queue_routing_warn";

  private static final String INTO_ROUTING_NAME= "info";
  private static final String WARN_ROUTING_NAME= "warn";
  private static final String ERROR_ROUTING_NAME= "error";


  public static void main(String[] args) throws IOException, TimeoutException {
    // 獲取連線
    Connection connection = ConnectionUtils.getConnection();

    // 開啟通道
    Channel channel = connection.createChannel();

    // 申明要消費的佇列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    // 繫結佇列到交換機
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, WARN_ROUTING_NAME);
    //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, INTO_ROUTING_NAME);
    //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ERROR_ROUTING_NAME);

    // 這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說，在接收到該Consumer的ack前，他它不會將新的Message分發給它。
    channel.basicQos(1);

    // 建立一個回撥的消費者處理類
    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 接收到的訊息
        String message = new String(body);
        System.out.println("WarnReceive '" + message + "' , 任務處理中");

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        } finally {
          System.out.println(" WarnReceive done ");
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

    // 消費訊息
    channel.basicConsume(QUEUE_NAME, false, consumer);

  }
}

4. 測試

提前在RabbitMQ Management建立一個direct交換機，或者先執行一次生產者（執行時會判斷交換機是否存在，不存在則建立交換機），這樣保證交換機存在，不然直接啟動消費者會提示交換機不存在。

注意點

如果在沒有佇列繫結在交換機上面時，往交換機發送訊息會丟失，之後繫結在交換機上面的佇列接收不到之前的訊息，也就是先執行第一次傳送，建立了交換機，但是還沒有佇列繫結在交換機上面，如果這次傳送的訊息就會丟失。

然後再啟動3上消費者,最後在啟動生產者.

執行結果 :

5. 多繫結情況

5.1 同一消費者繫結佇列多個路由鍵

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warn");

在`warn消費者中佇列繫結多個路由鍵 :

package com.makesailing.neo.consumer;

import com.makesailing.neo.utils.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * # 警告日誌 消費者
 *
 * @Author: jamie.li
 * @Date: Created in  2018/9/16 23:09
 */
public class WarnReceive {
  private static final String EXCHANGE_NAME = "test_exchange_direct";
  // info日誌佇列
  private static final String QUEUE_NAME = "test_queue_routing_warn";

  private static final String INTO_ROUTING_NAME= "info";
  private static final String WARN_ROUTING_NAME= "warn";
  private static final String ERROR_ROUTING_NAME= "error";


  public static void main(String[] args) throws IOException, TimeoutException {
    // 獲取連線
    Connection connection = ConnectionUtils.getConnection();

    // 開啟通道
    Channel channel = connection.createChannel();

    // 申明要消費的佇列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    // 繫結佇列到交換機
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, WARN_ROUTING_NAME);
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, INTO_ROUTING_NAME);
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ERROR_ROUTING_NAME);

    // 這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說，在接收到該Consumer的ack前，他它不會將新的Message分發給它。
    channel.basicQos(1);

    // 建立一個回撥的消費者處理類
    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 接收到的訊息
        String message = new String(body);
        System.out.println("WarnReceive '" + message + "' , 任務處理中");

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        } finally {
          System.out.println(" WarnReceive done ");
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

    // 消費訊息
    channel.basicConsume(QUEUE_NAME, false, consumer);

  }
}

執行結果:

如果一個消費者綁定了這3個路由鍵,那麼當生產者傳送其中一個路由鍵訊息時,該消費者都能接收到訊息.

5.2 多個消費者繫結相同的路由鍵

即消費者1繫結info,消費者2繫結 info、error .

那麼生產者傳送info路由鍵訊息時,消費者1 、2都能接收到訊息,傳送error路由鍵訊息時,只有消費者2能接收到訊息.

RabbitMQ實戰教程(五)-路由模式

1. 路由模式跟訂閱模式類似,只不過在訂閱模式的基礎上加上路由,訂閱模式是分發到所有繫結到該交換機的佇列,路由模式只分發到繫結在該交換機上面指定的路由鍵佇列. 2. Direct 直接交換機直連交換機(Direct Exchange)是一種帶路由功能的

(五) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ異常處理

RabbitMQ異常處理使用JAVA客戶端整合RabbitMQ進行的許多操作都會丟擲異常，我們可以自定義異常處理器進行處理，比如我們希望在RabbitMQ消費訊息失敗時記錄一條日誌，又或者在訊息消費失敗時傳送一則通知等操作 RabbitMQ Jav

RabbitMQ 一二事4 - 路由模式介紹

http dia cluster iis line max har 模式 abr %E5%9C%A8%20Windows%20%E4%B8%8A%E6%B5%8B%E8%AF%95%20Redis%20Cluster%E7%9A%84%E9%9B%86%E7%BE%A4%E

RabbitMQ實戰教程-鍾林森-專題視訊課程

RabbitMQ實戰教程—648人已學習課程介紹本課程適用於RabbitMQ實戰者以及SpringBoot整合RabbitMQ實戰需求者。課程將首先帶領大家拜讀RabbitMQ官

(九) RabbitMQ實戰教程(面向Java開發人員)之SpringBoot整合RabbitMQ

SpringBoot整合RabbitMQ 使用SpringBoot整合RabbitMQ非常簡單，它極大程度的簡化了開發成本，使用SpringBoot整合RabbitMQ需匯入如下依賴 <parent> <groupId>o

(四) RabbitMQ實戰教程(面向Java開發人員)之@RabbitListener訊息消費

使用RabbitListener註解進行訊息消費在前一篇部落格中我們往MessageListenerContainer設定了MessageListener進行訊息的消費，本篇部落格將介紹一種更為簡單的訊息消費方式:使用@RabbitListener註解方式。

(七) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ常用屬性詳解

RabbitMQ常用屬性詳解 Alternate Exchange Alternate Exchange簡稱AE，當訊息不能被正確路由時，如果交換機設定了AE則訊息會被投遞到AE中，如果存在AE鏈則會按此繼續投遞，直到訊息被正確路由或AE鏈結束訊息被

(三) RabbitMQ實戰教程(面向Java開發人員)之Spring整合RabbitMQ

Spring整合RabbitMQ Spring AMQP是對AMQP協議的抽象和封裝，從官方網站上得知它是由兩個專案組成的(spring-amqp和spring-rabbit)。在使用Spring整合RabbitMQ時我們主要關注三個核心介面(MessageL

(八) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ實現非同步RPC

RabbitMQ實現非同步RPC RabbitMQ Java Client 服務端步驟: 1.服務端監聽一個佇列，監聽客戶端傳送過來的訊息 2.收到訊息之後呼叫RPC服務得到呼叫結果 3.從訊息屬性中獲取reply_to，correlation_i

(二) RabbitMQ實戰教程(面向Java開發人員)之Rabbit Java Client

RabbitMQ Java Client 在介紹完RabbitMQ基本概念後，我們使用JAVA程式碼來模擬一套生產者和消費者的模型，Talk is cheap 直接上程式碼了。使用Java Client整合RabbitMQ需要在pom.xml中匯入如下依賴

(六) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ訊息的可靠性

訊息可靠性在專案中使用RabbitMQ時，我們可能會遇到這樣的問題:如一個訂單系統當用戶付款成功時我們往訊息中介軟體新增一條記錄期望訊息消費者修改訂單狀態，但是最終實際訂單狀態並沒有被修改成功。遇到這種問題我們排查的思路如下: 1.訊息是否已經成功傳送

RabbitMQ實戰教程(十二):訊息佇列的應用場景

這是網上的一篇教程寫的很好，不知原作者是誰，沒法註明出處，我看的時候也是別人轉載的，這裡就註明一下那篇轉載的地址：http://blog.csdn.net/cws1214/article/details/52922267 訊息佇列中介軟體是分散式系統中

(一) RabbitMQ實戰教程(面向Java開發人員)之RabbitMQ相關概念介紹

前言因為專案組需要對RabbitMQ的使用進行優化，所以系統化的學習了RabbitMQ，寫下本系列部落格的目的是希望能幫助到後續使用RabbitMQ的同學少走彎路，在閱讀本部落格前我希望您已經對RabbitMQ有基本的瞭解。本篇部落格主要面向JAVA開發人員

python使用rabbitMQ介紹四（路由模式）

一、模式介紹路由模式，與釋出-訂閱模式一樣，訊息傳送到exchange中，消費者把佇列繫結到exchange上。這種模式在exchange上新增添加了一個路由鍵（routing-key），生產者釋出訊息的時候新增路由鍵（routing-key），消費者繫結佇列到交換機時新增鍵值（routing-key

RabbitMQ 實戰教程（一）

MQ是消費-生產者模型的一個典型的代表，一端往訊息佇列中不斷寫入訊息，而另一端則可以讀取或者訂閱佇列中的訊息。RabbitMQ是資訊傳輸的中間者。本質上，他從生產者接收訊息，轉發這些訊息給消費者。換句話說，RabbitMQ能夠按根據你指定的規則進行訊息轉發、緩衝、和持久化。

Jasperreports6.3.1+Jaspersoft studio6.3.1進行報表開發實戰教程(五)

用實體類實現報表資料填充建立一個實體類上一節已經介紹瞭如何傳入引數填充報表，這一節要用另一個數據源來填充報表，也就是上一節介紹到的jasper.fill()方法的第三個引數JRDataSource，JavaBeanColletion可以用來生成一個JRDataSo

Spring Cloud 入門教程(五): 路由閘道器(zuul) (Greenwich.RELEASE)

一、準備工作啟動前面的專案：erurekaserver，eurekaclient1，service-feign，service

vue教程(五)--路由router介紹

一、html頁面中如何使用 1、引入 vue-router.js 2、安裝外掛 Vue.use(VueRouter) 3、建立路由物件　　var router = new VueRouter({ 　　　　// 4、配置路由　　　　routes:[{ 　　　　　　path:'/login',

RabbitMQ五種訊息佇列學習（五）--路由模式

RabbitMQ五種訊息佇列學習（五）–路由模式標籤（空格分隔）： RabbitMQ 釋出訂閱模式只是利用路由這個功能，將訊息全部分發給所有與路由繫結的佇列中，不能對不同的訊息進行選擇性的分發。模型結構路由根據Routing Key進行

Vue2+VueRouter2+webpack 構建項目實戰（五）：配置子路由

dex log fault 地址數據 from 插入接口 content 前言通過前面幾章的實戰，我們已經順利的構建項目，並且從API接口獲取到數據並且渲染出來了。制作更多的頁面，更復雜的應用，就是各位自己根據自己的項目去調整的事情了。本章講一下如何配置子路由，

RabbitMQ實戰教程(五)-路由模式

1. 路由模式

2. Direct 直接交換機

3. 程式碼演示

3.1 工具類

3.2 日誌生產者

3.3 error日誌消費者

3.4 info日誌消費者

3.5 warn日誌消費者

4. 測試

5. 多繫結情況

5.1 同一消費者繫結佇列多個路由鍵

5.2 多個消費者繫結相同的路由鍵

相關推薦