2. 程式人生 > >rabbitmq--topic主題路由模型

rabbitmq--topic主題路由模型

阿新 • • 發佈：2018-11-19

上一篇，我們介紹了直接路由模型，這一篇我們介紹下主題路由模型。

就如同上一篇所說，直接路由模型和主題路由模型非常相似，它們唯一區別在於：直接路由模型對於路由key是精確匹配的，而主題路由模型對於路由key是模糊匹配的(使用萬用字元 )。

因此當我們瞭解上一篇中的直接路由模型，也不難理解本篇中的主題路由模型了。如圖：

那麼topic中的路由key的匹配規則是什麼呢？ 

topic模式中路由key中存在兩種萬用字元  * 和 # 

  • * (star) can substitute for exactly one word.       *匹配明確的一個單詞
  • # (hash) can substitute for zero or more words.    #匹配0到多個單詞

 

topic模式中若路由key為"#" ，則表示對所有佇列感興趣 ，相當於fanout 

topic模式中若路由key中不含有*和#字元時，則表示只對精確匹配的路由key佇列感興趣，則相對於routing 
 

示例程式碼：

生產者 

package com.tingcream.rabbitmq.topic;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class TopicLogProducer {
	
	private static String EXCHANGE_NAME="topic_logs";
	
	public static void main(String[] args) throws Exception {
		
		
		 ConnectionFactory factory = new ConnectionFactory();
         
			//主機  埠  vhost 使用者名稱 密碼
			factory.setHost("192.168.9.102");
			factory.setUsername("rabbitmq");
			factory.setPassword("rabbitmq123");
			factory.setPort(AMQP.PROTOCOL.PORT);
			factory.setVirtualHost("/");
         
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        

//        lazy.black.elephant    懶惰的黑色的象
//        lazy.white.sheep       懶惰的白色的羊
//        quick.black.horse      快速的黑色的馬
//        quick.red.horse        快速的紅色的馬
        
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");  
        
        for (int i=0;i<10;i++){
            String message="你好,這是lazy.black.elephant級別訊息 "+i;
             channel.basicPublish(EXCHANGE_NAME,"lazy.black.elephant",null,message.getBytes());
             System.out.println("TopicLogProducer Send: " + message );
        }
        for (int i=0;i<10;i++){
        	String message="你好,這是lazy.white.sheep級別訊息 "+i;
        	channel.basicPublish(EXCHANGE_NAME,"lazy.white.sheep",null,message.getBytes());
        	System.out.println("TopicLogProducer Send: " + message );
        }
        for (int i=0;i<10;i++){
        	String message="你好,這是quick.black.horse級別訊息 "+i;
        	channel.basicPublish(EXCHANGE_NAME,"quick.black.horse",null,message.getBytes());
        	System.out.println("TopicLogProducer Send: " + message );
        }
        for (int i=0;i<10;i++){
        	String message="你好,這是quick.red.horse級別訊息 "+i;
        	channel.basicPublish(EXCHANGE_NAME,"quick.red.horse",null,message.getBytes());
        	System.out.println("TopicLogProducer Send: " + message );
        }
        channel.close();
        connection.close();
    }

}

消費者A

package com.tingcream.rabbitmq.topic;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class TopicLogReceiverA {
	private static    String EXCHANGE_NAME="topic_logs";
	
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory factory = new ConnectionFactory();
        
		//主機  埠  vhost 使用者名稱 密碼
		factory.setHost("192.168.9.102");
		factory.setUsername("rabbitmq");
		factory.setPassword("rabbitmq123");
		factory.setPort(AMQP.PROTOCOL.PORT);
		factory.setVirtualHost("/");
	    Connection connection=factory.newConnection();
	    Channel channel=connection.createChannel();
	    
	    
	    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
	    
	    //獲取一個隨機的佇列名稱   
	    String queueName = channel.queueDeclare().getQueue();
	    
	    
	    
	    
	    //  繫結routingKey "#" 表示對所有佇列感興趣 ，相當於fanout 
	    //  繫結routingKey 若不含"#"和"*",表示只對精確匹配的路由key佇列感興趣，則相對於routing 
	    
	    
	   // channel.queueBind(queue, exchange, routingKey)
	    channel.queueBind(queueName, EXCHANGE_NAME, "*.black.*"); 
	    System.out.println("TopicLogReceiverA Waiting for messages");
	    
	      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, "UTF-8");
	          System.out.println( "TopicLogReceiverA接收到訊息：" + message );   
	        }
	      };
	      channel.basicConsume(queueName, true, consumer);
	}

}

消費者B

 

package com.tingcream.rabbitmq.topic;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class TopicLogReceiverB {
	private static    String EXCHANGE_NAME="topic_logs";
	
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory factory = new ConnectionFactory();
        
		//主機  埠  vhost 使用者名稱 密碼
		factory.setHost("192.168.9.102");
		factory.setUsername("rabbitmq");
		factory.setPassword("rabbitmq123");
		factory.setPort(AMQP.PROTOCOL.PORT);
		factory.setVirtualHost("/");
	    Connection connection=factory.newConnection();
	    Channel channel=connection.createChannel();
	    
	    
	    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
	    
	    //獲取一個隨機的佇列名稱   
	    String queueName = channel.queueDeclare().getQueue();
	    
	    
	   // channel.queueBind(queue, exchange, routingKey)  //bind方法可以呼叫多次，繫結多個
	    channel.queueBind(queueName, EXCHANGE_NAME, "quick.#"); 
	    channel.queueBind(queueName, EXCHANGE_NAME, "*.white.*"); 
	    
	    
	    
	    System.out.println("TopicLogReceiverB Waiting for messages");
	    
	      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, "UTF-8");
	          System.out.println( "TopicLogReceiverB接收到訊息：" + message );   
	        }
	      };
	      channel.basicConsume(queueName, true, consumer);
	}

}

 

依次啟動消費者A、B和生產者。

消費者A輸出

TopicLogReceiverA Waiting for messages
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 0
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 1
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 2
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 3
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 4
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 5
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 6
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 7
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 8
TopicLogReceiverA接收到訊息：你好,這是lazy.black.elephant級別訊息 9
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 0
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 1
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 2
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 3
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 4
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 5
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 6
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 7
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 8
TopicLogReceiverA接收到訊息：你好,這是quick.black.horse級別訊息 9

消費者B輸出：

TopicLogReceiverB Waiting for messages
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 0
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 1
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 2
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 3
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 4
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 5
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 6
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 7
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 8
TopicLogReceiverB接收到訊息：你好,這是lazy.white.sheep級別訊息 9
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 0
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 1
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 2
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 3
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 4
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 5
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 6
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 7
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 8
TopicLogReceiverB接收到訊息：你好,這是quick.black.horse級別訊息 9
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 0
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 1
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 2
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 3
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 4
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 5
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 6
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 7
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 8
TopicLogReceiverB接收到訊息：你好,這是quick.red.horse級別訊息 9
 

管理後臺中 exchange標籤項中多出了一個topic交換機，名叫 topic_logs ，這就是我們剛宣告的topic型別交換機。

 

相關推薦

rabbitmq--topic主題路由模型

上一篇，我們介紹了直接路由模型，這一篇我們介紹下主題路由模型。 就如同上一篇所說，直接路由模型和主題路由模型非常相似，它們唯一區別在於：直接路由模型對於路由key是精確匹配的，而主題路由模型對於路由key是模糊匹配的(使用萬用字元 )。 因此當我們瞭解上一篇中的直接路由模型，也

rabbitmq--routing直接路由模型

rabbitmq中直接路由模型屬於釋出-訂閱模型中的一種，它是採用direct交換機，通過路由key精確匹配。   直接路由模型和主題路由模型非常相似，它們唯一區別在於：直接路由模型對於路由key是精確匹配的，而主題路由模型對於路由key是模糊匹配的(使用萬用字元 )

rabbitmq-routing直接路由模型

rabbitmq中直接路由模型屬於釋出-訂閱模型中的一種，它是採用direct交換機，通過路由key精確匹配。 直接路由模型和主題路由模型非常相似，它們唯一區別在於：直接路由模型對於路由key是精確匹配的，而主題路由模型對於路由key是模糊匹配的(使用萬用字元 )。

（八）RabbitMQ訊息佇列-通過Topic主題模式分發訊息

前兩章我們講了RabbitMQ的direct模式和fanout模式，本章介紹topic主題模式的應用。如果對direct模式下通過routingkey來匹配訊息的模式已經有一定了解那fanout也很好理解。簡單的可以理解成direct是通過routingkey精準匹配的，而topic是通過r

RabbitMQ3.7.2入門到進階之Topic主題模型

主題模式 生產者 public class Send { private final static String EXCHANGE_NAME = "test_exchang

RabbitMQ Topic exchange

logs url 創建 routing 模式 get 動態 生產 分享 RabbitMQ Topic exchange Topic exchange topic與之前的每個類型都不同（ps:廢話每個都是不同的）。Topic解決了我們另一個需求。舉個例子，有一個做資訊

Python-RabbitMQ(簡單發送模型)

tar 開始 AS lar exchange exc MQ port 本機 RabbitMQ需要 erlang 和pika 發送端：生產者 import pika connection = pika.BlockingConnection( pika.Connec

SpringCloud Stream生產者配置RabbitMq的動態路由鍵

在寫這個文章前不得不吐槽目前國內一些blog的文章，盡是些複製貼上的文章，提到點上但沒任何的深入和例子。......... 經過測試下來總結一下RabbitMQ的Exchange的特性： 1、direct 生產者可以指定路由鍵，消費者可以指定路由鍵，但不能講路由鍵設定為#(全部)。 2、topic

4.RabbitMQ Topic/Publish

注意 消費者需要先啟動，不然會有訊息黑洞，即消失丟失現象。 package com.study.soufang.rabbit.a001.topicAndPublish; public class ConstantOfTopic {     publi

金融主題域模型

大資料平臺建設後，資料經過初步的清洗、簡單加工進入到貼源層，面對這些資料，如何在貼源層之上的基礎層構建主題模型是需要思考的一個課題。清晰的資料結構可以在使用表的時候能更方便地定位和理解，規範資料分層，開發一些通用的中間層資料，比如彙總層資料，滿足不同業務取數要求

RabbitMQ : Topic 交換器(規則匹配)

Topic交換器型別與Direct類似，都是通過路由鍵來匹配訊息佇列，不同之處是Direct是通過完全匹配路由鍵的名稱對應，而Topic型別支援規則匹配，也即支援萬用字元的方式。 下面通過不同的服務對不同級別的日誌傳送到不同規則的訊息佇列 一、建立接收者（消費者）專案

雲服務的抽象路由模型

雲服務路由的訴求 在PaaS或SaaS領域內，雲服務是PaaS與SaaS中第二個S（Service）的實現者，一個看似簡單的雲服務，其實現上通常是很複雜的，且並不是獨立的。 典型的如亞馬遜的S3服務，S3本身是一個跨地域對全球提供服務的服務，S3服務內部之間需要互相通訊交換資料與指令，同時S3又依賴

laravel 路由模型綁定

情況下 存在 orf 請求 fun 就是 控制 size 新增 我們在使用路由的時候一個很常見的使用場景就是根據資源 ID 查詢資源信息： Route::get(‘task/{id}‘, function ($id) { $task = \App\Models\Ta

RabbitMQ入門：路由(Routing)

在上一篇部落格《RabbitMQ入門：釋出/訂閱（Publish/Subscribe）》中，我們認識了fanout型別的exchange，它是一種通過廣播方式傳送訊息的路由器，所有和exchange建立的繫結關係的佇列都會接收到訊息。但是有一些場景只需要訂閱到一部分訊息，這個時候就不能使用fanout 型別的

ActiveMQ學習總結——（三）Topic主題模式示例

和佇列模式相似，分別編寫生產者和訂閱者。 生產者： package com.jms.topic; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.j

資料倉庫專題20-案例篇：電商領域資料主題域模型設計v0.1(改進意見徵集中）

一、電商分類（平臺+自營+複合）  （1）平臺型電商：淘寶+天貓+百度Mall等；  （2）自營型電商：         2.1 綜合型：京東（早期）+噹噹（早期）；         2.2 垂直型：好像這種型別越來越少了；  （3）複合型電商（平臺+自營）：京東+噹噹+

訊息佇列MQ實踐----實現Queue(佇列訊息)和Topic(主題訊息)兩種模式

之前有篇檔案介紹了生產消費者模式(http://blog.csdn.net/canot/article/details/51541920 )，當時是通過BlockingQueue阻塞佇列來實現,以及在Redis中使用pub/sub模式(http://blog.csdn.ne

LDA（文件主題生成模型）--零基礎（一）

本文目的在於用最簡單的語言，給大家介紹LDA的原理以及使用方法。 首先，我們這裡有M篇文件，每篇文件有個單詞，如果文件是英文，就是每個單獨的單詞；如果是中文，則需要分詞處理，分詞處理又需要用已有的分詞的演算法，比較複雜，所以我們這裡先只考慮英文。我們現在想知道，每篇文章，

RabbitMQ使用多路由，多佇列來破除流控

流控機制是我們在使用RabbitMQ最頭疼的問題，一旦併發激增時，消費者消費佇列訊息就像滴水一樣慢。 現在我們下單後，需要給通知

RabbitMQ入門：主題路由器(Topic Exchange)

AI orange topic 都是 erro col host nfa 匹配 上一篇博文中，我們使用direct exchange 代替了fanout exchange，這次我們來看下topic exchange。 一、Topic Exchange介紹 topic e