RabbitMQ入门:主题路由器(Topic Exchange)[通俗易懂]

上一篇博文中,我们使用directexchange代替了fanoutexchange,这次我们来看下topicexchange。一、TopicExchange介绍topicexchan

大家好,又见面了,我是全栈君。

上一篇博文中,我们使用direct exchange 代替了fanout exchange,这次我们来看下topic exchange。

一、Topic Exchange介绍

topic exchange和direct exchange类似,都是通过routing key和binding key进行匹配,不同的是topic exchange可以为routing key设置多重标准。

direct路由器类似于sql语句中的精确查询;topic 路由器有点类似于sql语句中的模糊查询。

还记得吗?我们在《RabbitMQ入门:发布/订阅(Publish/Subscribe)》中对exchange的分类进行过介绍:

Direct:完全根据key进行投递的,例如,绑定时设置了routing key为”abc”,那么客户端提交的消息,只有设置了key为”abc”的才会投递到队列。
Topic:对key进行模式匹配后进行投递,符号”#”匹配一个或多个词,符号”*”匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”,”abc.*”只匹配”abc.def”。
Fanout:不需要key,它采取广播模式,一个消息进来时,投递到与该交换机绑定的所有队列。
Headers:我们可以不考虑它。

下面是官网给出的工作模型(P代表生产者,X代表exhange,红色的Q代表队列,C代表消费者):

RabbitMQ入门:主题路由器(Topic Exchange)[通俗易懂]

我们来分析下这个模型。

它发送的消息是用来描述动物的。路由键有三个单词:<speed>.<color>.<species>,第一个单词描述了速度,第二个描述了颜色,第三个描述了物种。
有三个绑定键,Q1绑定键为*.orange.*(关注所有颜色为orange的动物); Q2的绑定键有两个,分别是*.*.rabbit(关注所有的兔子)和lazy.#(关注所有速度为lazy的动物)。

因此,路由键为quick.orange.rabbit的消息将发送到Q1和Q2,路由键为quick.orange.fox的消息将发送到Q1,路由键为lazy.brown.fox的消息将发送到Q2。路由键为lazy.pink.rabbit的消息将发送到Q2,但是注意,它只会到达Q2一次,尽管它匹配了两个绑定键。路由键为quick.brown.fox的消息因为不和任意的绑定键匹配,所以将会被丢弃。

如果有人手一抖发了个lazy.orange.male.rabbit这种四个单词的,这个怎么办呢? 由于它和lazy.#匹配,因此将发送到Q2。

二、代码示例

接下来我们看下代码

  1. 生产者
    public class LogTopicSender {
        // exchange名字
        public static String EXCHANGE_NAME = "topicExchange";
    
        public static void main(String[] args) {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = null;
            Channel channel = null;
            try {
                // 1.创建连接和通道
                connection = factory.newConnection();
                channel = connection.createChannel();
    
                // 2.为通道声明topic类型的exchange
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
                
                // 3.发送消息到指定的exchange,队列指定为空,由exchange根据情况判断需要发送到哪些队列
                String routingKey = "info";
    //            String routingKey = "log4j.error";
    //            String routingKey = "logback.error";
    //            String routingKey = "log4j.warn";
                String msg = " hello rabbitmq, I am " + routingKey;
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
                System.out.println("product send a msg: " + msg);
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            } finally {
                // 4.关闭连接
    
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    }

     

  2. 消费者
    public class LogTopicReciver {
    
        public static void main(String[] args) {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = null;
            Channel channel = null;
            try {
                // 1.创建连接和通道
                connection = factory.newConnection();
                channel = connection.createChannel();
    
                // 2.为通道声明topic类型的exchange
                channel.exchangeDeclare(LogTopicSender.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
                // 3.创建随机名字的队列
                String queueName = channel.queueDeclare().getQueue();
    
                // 4.建立exchange和队列的绑定关系
                String[] bindingKeys = { "#" };
    //            String[] bindingKeys = { "log4j.*", "#.error" };
    //            String[] bindingKeys = { "*.error" };
    //            String[] bindingKeys = { "log4j.warn" };
                for (int i = 0; i < bindingKeys.length; i++) {
                    channel.queueBind(queueName, LogTopicSender.EXCHANGE_NAME, bindingKeys[i]);
                    System.out.println(" **** LogTopicReciver keep alive ,waiting for " + bindingKeys[i]);
                }
    
                // 5.通过回调生成消费者并进行监听
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                            com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) throws IOException {
    
                        // 获取消息内容然后处理
                        String msg = new String(body, "UTF-8");
                        System.out.println("*********** LogTopicReciver" + " get message :[" + msg + "]");
                    }
                };
                // 6.消费消息
                channel.basicConsume(queueName, true, consumer);
    
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }

     

  3. 启动消费者,作为消费者1
  4. 分别将String[] bindingKeys = { “#” };改为String[] bindingKeys = { “log4j.*”, “#.error” };/String[] bindingKeys = { “*.error” };/String[] bindingKeys = { “log4j.warn” };,然后启动作为消费者2、消费者3、消费者4
  5. 启动4次生产者,routing key分别为String routingKey = “info”;、String routingKey = “log4j.error”;、String routingKey = “logback.error”;、String routingKey = “log4j.warn”;
  6. 观察控制台log
    生产者:
    product send a msg:  hello rabbitmq, I am info
    product send a msg:  hello rabbitmq, I am log4j.error
    product send a msg:  hello rabbitmq, I am logback.error
    product send a msg:  hello rabbitmq, I am log4j.warn
    
    消费者1:
     **** LogTopicReciver keep alive ,waiting for #
    *********** LogTopicReciver get message :[ hello rabbitmq, I am info]
    *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.error]
    *********** LogTopicReciver get message :[ hello rabbitmq, I am logback.error]
    *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.warn]
    
    消费者2:
     **** LogTopicReciver keep alive ,waiting for log4j.*
     **** LogTopicReciver keep alive ,waiting for #.error
    *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.error]
    *********** LogTopicReciver get message :[ hello rabbitmq, I am logback.error]
    *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.warn] 消费者3: **** LogTopicReciver keep alive ,waiting for *.error *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.error] *********** LogTopicReciver get message :[ hello rabbitmq, I am logback.error] 消费者4: **** LogTopicReciver keep alive ,waiting for log4j.warn *********** LogTopicReciver get message :[ hello rabbitmq, I am log4j.warn]

     

  7. 观察RabbitMQ管理页面RabbitMQ入门:主题路由器(Topic Exchange)[通俗易懂]

    RabbitMQ入门:主题路由器(Topic Exchange)[通俗易懂]

    RabbitMQ入门:主题路由器(Topic Exchange)[通俗易懂]

     

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/120895.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 数字转罗马字符串_整数转罗马数字 java

    数字转罗马字符串_整数转罗马数字 java//数字转罗马字符串//VC2008通过#include”stdafx.h”#include//数字转罗马字符串//需要保证strData空间足够//范围:1~3999voidNumberToString(intnValue,char*strData){ constchar*s1000[4]={“”,”M”,”MM”,”MMM”}; c

  • 手机版mt4平台下载下载手机版mt4_手机mt4平台下载安卓

    手机版mt4平台下载下载手机版mt4_手机mt4平台下载安卓MetaTrader4(缩写MT4)是由俄罗斯迈达克软件公司(英文名称MetaQuotesSoftwareCorp)发布的一款交易软件,可同时在电脑端和手机端使用。但是,正版手机端mt4一般要从谷歌商店下载,而GooglePlay的下载地址经常打不开。本人使用mtr软件多年,在使用方面有一些经验,现将2020年版手机mt4下载地址分享给各位网友,愿您交易如索罗斯一般行如流水!mt4下载地址:https://t00y.com/file/29212124-458672692提取码:pj265(

  • Mysql中行转列和列转行

    Mysql中行转列和列转行一、行转列即将原本同一列下多行的不同内容作为多个字段,输出对应内容。建表语句DROPTABLEIFEXISTStb_score;CREATETABLEtb_score(idINT(11)NOTNULLauto_increment,useridVARCHAR(20)NOTNULLCOMMENT’用户id’,subject…

  • 传统请求风格 VS RestFul 风格

    传统请求风格 VS RestFul 风格RestFul风格概念Restful就是一个资源定位及资源操作的风格。不是标准也不是协议,只是一种风格。基于这个风格设计的软件可以更简洁,更有层次,更易于实现缓存等机制。功能资源:互联网所有的事物都可以被抽象为资源资源操作:使用POST、DELETE、PUT、GET,使用不同方法对资源进行操作。分别对应添加、删除、修改、查询。传统方式操作资源:通过不同的参数来实现不同的效果!方法单一,post和get​ http://127.0.0.1/item/queryItem.actio

  • 浅谈一个网页打开的全过程(涉及DNS、CDN、Nginx负载均衡等)

    浅谈一个网页打开的全过程(涉及DNS、CDN、Nginx负载均衡等)

    2021年10月31日
  • 数据不平衡之SMOTE算法

    数据不平衡之SMOTE算法在企业的数据分析中,很少会遇到正负样本数据比例平衡的状况。通常情况是,绝大多数为正样本,而只有极少数(几个或者十几个)负样本。在这种情况下,不论是用LR,SVM或者基于提升方法的随机森林,直接用该数据集进行学习的效果都不会太好,原因是这些方法的学习结果都会偏向于样本较多的一类。另一个方面,对学习结果进行评估时,假如正样本占95%,负样本仅占5%,这样甚至不需要学习,直接把所有新样本预测为正,准确率

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号