【RabbitMQ】——三种Exchange模式(Fanout、Direct、Topic)

简介: 【RabbitMQ】——三种Exchange模式(Fanout、Direct、Topic)

引言

 

在上一篇博客中介绍了RabbitMQ中两种队列模式,下面在改博文中将介绍比较常用的三种Exchange模式。


第一种、Fanout Exchange


 这种Fanout模式不处理路由键,你需要简单的讲队列绑定到exchange上,一个发送到exchange的消息都会被转发到与该exchange绑定的所有队列上。很像广播子网,每台子网内的主机都获得了一份复制的消息。Fanout类型的Exchange转发消息是最快的。下面看一张图;

20170512103945228.png

从上面这张图中我们发现,在这种模式中增加了Exchange的概念,这就说生产者会创建一个Exchange,并且将消息发送到exhange中,其余的事情不在关心,并且这种模式中,我们会发现拥有多个消费者,并且每个消费者都有一个独立的消息队列。 每个队列都需要绑定到exchange上面。生产者发送的消息,经过exchange到达队列,从而实现一个消息被多个消费者获取的目的。下面看一个比较简单的图:


20170512104521047.png

在这种模式中我们需要注意的,如果将消息发送到一个没有队列绑定的exchange上面,那么该消息将会丢失,这是因为在rabbitmq中exchange是不具备存储消息的能力,只有队具备存储消息的能力。

 

示例代码:生产者

package cn.itcast.rabbitmq.ps;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
  private final static String EXCHANGE_NAME = "test_exchange_fanout";
  public static void main(String[] argv) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明exchange 在这定义的交换机的类型是fanout 一共有三种类型的交换机
    channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    // 消息内容
    String message = "商品已经新增,id = 1000";
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");
    channel.close();
    connection.close();
  }
}

示例代码:消费者

package cn.itcast.rabbitmq.ps;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv {
    private final static String QUEUE_NAME = "test_queue_fanout_1";
    private final static String EXCHANGE_NAME = "test_exchange_fanout";
    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 监听队列,手动返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 前台系统: '" + message + "'");
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

消费者2代码和消费者1的代码相似,只是声明的队列名称不同,因为在这种模式中,每个消费者都拥有自己的队列。


第二种、Direct Exchange


 这种模式会处理路由键,也就是所谓的按需索取。需要将一个队列绑定到exchange上面,要求该消息与一个特定的路由键完全匹配。消息才会进入到该队列中。如果一个队列绑定到该交换机上要求路由键 “dog”,则只有被标记为“dog”的消息才被转发,不会转发dog.puppy,也不会转发dog.guard,只会转发dog。  


20170512105615019.png

这种模式就要去我们生产者在发送消息的时候,为此消息添加一个路由键,这样在消费者获取的时候才能达到“按需索取”。例如:将不同的级别的错误消息传递给不同的消费者,这就需要用到这种模式


20170512105930178.png

示例代码:生成者

package cn.itcast.rabbitmq.routing;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
  private final static String EXCHANGE_NAME = "test_exchange_direct";
  public static void main(String[] argv) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明exchange 和发布订阅模式相比 exchange模式为“direct”
    channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    // 消息内容
    String message = "删除商品, id = 1001";
    // 在路由模式中,我们在发布消息的时候需要指定路由key
    channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");
    channel.close();
    connection.close();
  }
}


示例代码:消费者1

package cn.itcast.rabbitmq.routing;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv {
  private final static String QUEUE_NAME = "test_queue_direct_1";
  private final static String EXCHANGE_NAME = "test_exchange_direct";
  public static void main(String[] argv) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    // 绑定队列到交换机 第三个参数的专业叫法:路由键
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    // 同一时刻服务器只会发一条消息给消费者
    channel.basicQos(1);
    // 定义队列的消费者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 监听队列,手动返回完成
    channel.basicConsume(QUEUE_NAME, false, consumer);
    // 获取消息
    while (true) {
      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      System.out.println(" 前台系统: '" + message + "'");
      Thread.sleep(10);
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
  }
}

消费者2:

package cn.itcast.rabbitmq.routing;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv2 {
    private final static String QUEUE_NAME = "test_queue_direct_2";
    private final static String EXCHANGE_NAME = "test_exchange_direct";
    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 监听队列,手动返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 搜索系统: '" + message + "'");
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}


从上面的代码中我们可以看到,两个不同的消费者在将队列绑定到exchange上面的时候,指定了不同的路由键。这时候我们如果更改生产者端的路由键的时候,则消息会路由到不同的消费者上面。如果生产者指定路由键为“insert”则只有消费者2获取,如果指定路由键为“update”则两个消费者都会获取。如果指定路由键为“abc”,则两个消费者都不会获取,从而导致消息进入”黑洞“。


 但是这种模式在一定程度上满足了我们的需求,但是在使用起来并不是非常的灵活,所以下面介绍一种更为灵活的模式。


第三种、Topic Exchange


 这种模式和Direct模式的原理是一样的,都是根据路由键进行消息的路由,但是这种支持路由键的模糊匹配,此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词,符号“*”匹配不多不少一个词。下面来看一张图:



20170512111129762.png


上面这张图根据颜色进行了路由键匹配的划分,下面四个队列中第一个红色队列,我们会发现它的路由键为“usa.#”这说明凡是路由键以“usa.”开头的消息都会被路由到这个队列中。第二个黄色队列,说明凡是以“.news”结束的路由键都会被路由到这个队列中。这样的模式是我们在使用过程中更加的灵活。


20170512111611031.png


示例代码:生产者


package cn.itcast.rabbitmq.topic;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息内容
        String message = "删除商品,id = 1001";
        channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

示例代码:消费者

package cn.itcast.rabbitmq.topic;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv {
    private final static String QUEUE_NAME = "test_queue_topic_1";
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 监听队列,手动返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 前台系统: '" + message + "'");
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}


消费者1是”按需索取“,所以他并不能使用通配符模式,而是使用完全匹配。

示例代码:消费者2

 

package cn.itcast.rabbitmq.topic;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv2 {
    private final static String QUEUE_NAME = "test_queue_topic_2";
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 监听队列,手动返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 搜索系统: '" + message + "'");
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

在消费者2中使用通配符模式,更加方便的编写代码,进行消息的路由,这中模式在消息类型比较负责,并且需要进行精细的处理的时候还是比较使用的。

 

三种队列对比

 

Fanout类型没有路由键的概念,只要队列绑定到了改exchange上面,就会接收到所有的消息。其余两种在此基础上面引进了路由键的概念,这样实现消费者”按需索取“的模式。

 

和前面的(简单模式和work模式)两种对比

 

前面的两种模式中 只有一个队列,消费者都是对一个队列进行消费的。同一个消息只能被一个消费者获取。而在exchange模式中,很好的实现了一个消息被多个消费者获取的需求。通过引入路由键从而更好的实现需求。


小结


 这几种模式是rabbitmq中比较重要的个知识点,小编在这总结记录一下,方便日后回顾。如果有理解错误的地方,欢迎拍砖

相关实践学习
消息队列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的 RPC 消息模式你会了吗?
【9月更文挑战第11天】RabbitMQ 的 RPC(远程过程调用)消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括:1) 客户端发送请求,创建回调队列并设置关联标识符;2) 服务器接收请求并发送响应至回调队列;3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序,但需根据需求调整优化。
|
2月前
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
172 7
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
|
25天前
|
消息中间件 负载均衡 算法
聊聊 RocketMQ中 Topic,Queue,Consumer,Consumer Group的关系
本文详细解析了RocketMQ中Topic、Queue、Consumer及Consumer Group之间的关系。文中通过图表展示了Topic可包含多个Queue,Queue分布在不同Broker上;Consumer组内多个消费者共享消息;并深入探讨了集群消费与广播消费模式下Queue与Consumer的关系,以及Rebalancing机制在实例增减时如何确保负载均衡。理解这些关系有助于更好地掌握RocketMQ的工作原理,提升系统运维效率。
52 2
|
3月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配:掌握消息路由的艺术!
【8月更文挑战第24天】在消息队列(MQ)体系中,交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器,它通过模式匹配实现消息的精准分发,适用于发布-订阅模式。不同于直接交换器和扇形交换器,`topic`交换器支持更复杂的路由策略,通过带有通配符(如 * 和 #)的模式字符串来定义队列与交换器间的绑定关系。
54 2
|
3月前
|
消息中间件 应用服务中间件 网络安全
rabbitMQ镜像模式搭建
rabbitMQ镜像模式搭建
|
3月前
|
消息中间件 Java Maven
RabbitMQ通配符模式
RabbitMQ通配符模式
57 0
|
1天前
|
消息中间件 存储 Kafka
MQ 消息队列核心原理,12 条最全面总结!
本文总结了消息队列的12个核心原理,涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。
|
5天前
|
消息中间件
解决方案 | 云消息队列RabbitMQ实践获奖名单公布!
云消息队列RabbitMQ实践获奖名单公布!
|
7天前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
25 3
|
12天前
|
消息中间件 存储 弹性计算
云消息队列RabbitMQ实践
云消息队列RabbitMQ实践

热门文章

最新文章