RabbitMQ实例教程:主题交换机

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介:

  前面的例子中,尽管我们使用了direct路由代替fanout路由解决了盲目广播的问题,但direct路由也有它的缺陷,他不能基于多个标准做路由转发。


  在上面的日志系统中,如果不仅想基于日志等级做订阅,也想根据日志的发生源做订阅该怎么处理呢?这时候你可能想到了unix系统工具中的syslog服务,它不仅基于日志等级(info/warn/crit...)进行路由转发,也会根据操作(auth/cron/kern...)做路由转发。


  如果是那样的话,日志系统就灵活多了,它不仅能够监听来自‘cron’的关键错误,也能监听来自'kern'的所有日志。其实主题交换机(topic exchange)就能解决这种问题。


  主题交换机(Topic exchange)


  主题交换机的路由代码不能是任意写的,必须是小树点分隔开的一组单词列表。这些单词可以随便写,但通常是与连接消息特征有关的单词。有效地路由代码应该是这样的“stock.usd.nyse”,“nyse.vmw”,“quick.orange.rabbit”。路由代码可以随便写,但是长度限制在255字节。


  注意,绑定代码也必须在同一个表单中。topic交换机与direct交换机类似-具有特定路由代码的消息会传送给所有匹配绑定代码的队列,但有两个特殊的绑定代码:


  * :它能替代一个单词 


  #:它能替代0或多个单词

wKiom1YekEyRndmFAABl8iEOXGg312.jpg


  该例子中,我们给所有的动物发送消息,符合由三个单词(第一个单词描述速度;第二个单词描述颜色;第三个单词描述物种)组成的路由代码将会发送消息:“<speed>.<colour>.<species>”。

   

  我们创建了三个绑定:Q1使用“*.orange.*”绑定,Q2使用“*.*.rabbit”和“lazy.#”绑定。这些绑定的意义如下:

  

  Q1描述了所有颜色为橙色的动物。

  

  Q2描述了是兔子的动物和懒惰的动物。

  

  这样,“quick.orange.rabbit”消息通过路由转发给Q1、Q2两个队列。"lazy.orange.elephant"消息也会转发给Q1、Q2两个队列。“quick.orange.fox”消息只会转发给Q1队列,"lazy.brown.fox"也只会转发给Q2队列。"lazy.pink.rabbit"会转发给Q2队列一次,尽管它匹配两个绑定。"quick.brown.fox"并不匹配任何一个队列就会被废弃。

  

  如果我们打破规则,每次只发一个或四个单词的话,如“orange”或”quick.orange.male.rabbit“,这些消息不匹配任何绑定,就会被废弃。但如果发送”lazy.orange.male.rabbit“这样的消息的话,由于它匹配最后的绑定仍会被转发到Q2队列中。

    

  主题交换机是一种非常强大的交换机,当它只绑定”#“时,它会接收所有的消息,与fanout交换机类似。当没有使用”*“和”#“符号时,主题交换机的作用等同与direct交换机。


  源代码


EmitLogTopic.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package  com.favccxx.favrabbit;
 
import  com.rabbitmq.client.Channel;
import  com.rabbitmq.client.Connection;
import  com.rabbitmq.client.ConnectionFactory;
 
public  class  EmitLogTopic {
 
     private  static  final  String EXCHANGE_NAME =  "topic_logs" ;
 
     public  static  void  main(String[] argv) {
         Connection connection =  null ;
         Channel channel =  null ;
         try  {
             ConnectionFactory factory =  new  ConnectionFactory();
             factory.setHost( "localhost" );
 
             connection = factory.newConnection();
             channel = connection.createChannel();
 
             channel.exchangeDeclare(EXCHANGE_NAME,  "topic" );
 
             String[] routingKeys = {  "fast.orange.duck" "slow.orange.fish" "grey.rabbit" "fast.black.rabbit" ,
                     "quick.white.rabbit" "lazy.dog" "lazy.black.pig"  };
             String[] messages = {  "Hello" "Guys" "Girls" "Babies"  };
 
             for  ( int  i =  0 ; i < routingKeys.length; i++) {
                 for  ( int  j =  0 ; j < messages.length; j++) {
                     channel.basicPublish(EXCHANGE_NAME, routingKeys[i],  null , messages[j].getBytes( "UTF-8" ));
                     System.out.println( " [x] Sent '"  + routingKeys[i] +  "':'"  + messages[j] +  "'" );
                 }
             }
         catch  (Exception e) {
             e.printStackTrace();
         finally  {
             if  (connection !=  null ) {
                 try  {
                     connection.close();
                 catch  (Exception ignore) {
                 }
             }
         }
     }
}


ReceiveLogsTopic.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package  com.favccxx.favrabbit;
 
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  ReceiveLogsTopic {
 
     private  static  final  String EXCHANGE_NAME =  "topic_logs" ;
 
     public  static  void  main(String[] argv)  throws  Exception {
         ConnectionFactory factory =  new  ConnectionFactory();
         factory.setHost( "localhost" );
         Connection connection = factory.newConnection();
         Channel channel = connection.createChannel();
 
         channel.exchangeDeclare(EXCHANGE_NAME,  "topic" );
         String queueName = channel.queueDeclare().getQueue();
 
         String[] bindingKeys = {  "*.orange.*" "*.*.rabbit" "lazy.#"  };
         for  ( final  String bindingKey : bindingKeys) {
             channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
             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( "["  + bindingKey +  "] Received message :'"  + message +  "' from routingKey : "  + envelope.getRoutingKey());
                 }
             };
             channel.basicConsume(queueName,  true , consumer);
         }
 
     }
}


运行消息发送器,在消息接收平台输出内容如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
[*.orange.*] Received message : 'Hello'  from routingKey : fast.orange.duck
[*.*.rabbit] Received message : 'Guys'  from routingKey : fast.orange.duck
[lazy. #] Received message :'Girls' from routingKey : fast.orange.duck
[*.orange.*] Received message : 'Babies'  from routingKey : fast.orange.duck
[*.*.rabbit] Received message : 'Hello'  from routingKey : slow.orange.fish
[lazy. #] Received message :'Guys' from routingKey : slow.orange.fish
[*.orange.*] Received message : 'Girls'  from routingKey : slow.orange.fish
[*.*.rabbit] Received message : 'Babies'  from routingKey : slow.orange.fish
[lazy. #] Received message :'Hello' from routingKey : fast.black.rabbit
[*.orange.*] Received message : 'Guys'  from routingKey : fast.black.rabbit
[*.*.rabbit] Received message : 'Girls'  from routingKey : fast.black.rabbit
[lazy. #] Received message :'Babies' from routingKey : fast.black.rabbit
[*.orange.*] Received message : 'Hello'  from routingKey : quick.white.rabbit
[*.*.rabbit] Received message : 'Guys'  from routingKey : quick.white.rabbit
[lazy. #] Received message :'Girls' from routingKey : quick.white.rabbit
[*.orange.*] Received message : 'Babies'  from routingKey : quick.white.rabbit
[*.*.rabbit] Received message : 'Hello'  from routingKey : lazy.dog
[lazy. #] Received message :'Guys' from routingKey : lazy.dog
[*.orange.*] Received message : 'Girls'  from routingKey : lazy.dog
[*.*.rabbit] Received message : 'Babies'  from routingKey : lazy.dog
[lazy. #] Received message :'Hello' from routingKey : lazy.black.pig
[*.orange.*] Received message : 'Guys'  from routingKey : lazy.black.pig
[*.*.rabbit] Received message : 'Girls'  from routingKey : lazy.black.pig
[lazy. #] Received message :'Babies' from routingKey : lazy.black.pig




本文转自 genuinecx 51CTO博客,原文链接:http://blog.51cto.com/favccxx/1703031,如需转载请自行联系原作者
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
240 7
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
|
6月前
|
消息中间件 Java RocketMQ
RocketMQ实战教程之RocketMQ安装
这是一篇关于RocketMQ安装的实战教程,主要介绍了在CentOS系统上使用传统安装和Docker两种方式安装RocketMQ。首先,系统需要是64位,并且已经安装了JDK 1.8。传统安装包括下载安装包,解压并启动NameServer和Broker。Docker安装则涉及安装docker和docker-compose,然后通过docker-compose.yaml文件配置并启动服务。教程还提供了启动命令和解决问题的提示。
|
1月前
|
消息中间件 存储 JSON
rabbitmq基础教程(ui,java,springamqp)
本文提供了RabbitMQ的基础教程,包括如何使用UI创建队列和交换机、Java代码操作RabbitMQ、Spring AMQP进行消息发送和接收,以及如何使用不同的交换机类型(fanout、direct、topic)进行消息路由。
25 0
rabbitmq基础教程(ui,java,springamqp)
|
2月前
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
87 0
|
2月前
|
消息中间件 存储
RabbitMQ-死信交换机和死信队列
死信队列和死信交换机是RabbitMQ提供的一个非常实用的功能,通过合理使用这一机制,可以大大增强系统的健壮性和可靠性。它们不仅能有效解决消息处理失败的情况,还能为系统的错误追踪、消息延迟处理等提供支持。在设计系统的消息体系时,合理规划和使用死信队列和死信交换机,将会为系统的稳定运行提供一个有力的
62 0
|
3月前
|
网络协议 物联网 测试技术
App Inventor 2 MQTT拓展入门(保姆级教程)
本文演示的是App和一个测试客户端进行消息交互的案例,实际应用中,我们的测试客户端可以看着是任意的、支持MQTT协议的硬件,通过订阅及发布消息,联网硬件与我们的App进行双向数据通信,以实现万物互联的智能控制效果。
202 2
|
3月前
|
消息中间件 监控 Ubuntu
RabbitMQ安装配置,超详细版教程
以上步骤为您提供了在Linux环境下安装RabbitMQ的详细过程。安装Erlang作为基础,然后通过添加官方源并安装RabbitMQ本身,最后对服务进行配置并启用Web管理界面。这些步骤操作简单直观,只需要跟随上述指南,即可在短时间内将RabbitMQ服务器运行起来,并进行进一步的配置和管理。不要忘记硬件和网络资源对性能的影响,确保RabbitMQ能够满足您的应用需求。
260 0
|
4月前
|
消息中间件 搜索推荐 RocketMQ
消息队列 MQ使用问题之如何将一个主题的多个分区分布到不同的Broker上
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件 存储 Apache
RocketMQ实战教程之常见概念和模型
Apache RocketMQ 实战教程介绍了其核心概念和模型。消息是基本的数据传输单元,主题是消息的分类容器,支持字节、数字和短划线命名,最长64个字符。消息类型包括普通、顺序、事务和定时/延时消息。消息队列是实际存储和传输消息的容器,是主题的分区。消费者分组是一组行为一致的消费者的逻辑集合,也有命名限制。此外,文档还提到了一些使用约束和建议,如主题和消费者组名的命名规则,消息大小限制,请求超时时间等。RocketMQ 提供了多种消息模型,包括发布/订阅模型,有助于理解和优化消息处理。
|
6月前
|
消息中间件 存储 Java
RocketMQ实战教程之NameServer与BrokerServer
这是一个关于RocketMQ实战教程的概要,主要讨论NameServer和BrokerServer的角色。NameServer负责管理所有BrokerServer,而BrokerServer存储和传输消息。生产者和消费者通过NameServer找到合适的Broker进行交互,不需要直接知道Broker的具体信息。工作流程包括生产者向NameServer查询后发送消息到Broker,以及消费者同样通过NameServer获取消息进行消费。这种设计类似于服务注册中心的概念,便于系统扩展和集群管理。