强烈建议你试试无所不能的chatGPT，快点击我

RabbitMQ】三种Exchange模式——订阅、路由、通配符模式

阅读量：5729 次

发布时间：2019-06-18

本文共 8599 字，大约阅读时间需要 28 分钟。

这篇博客介绍订阅、路由和通配符模式，之所以放在一起介绍，是因为这三种模式都是用了Exchange交换机，消息没有直接发送到队列，而是发送到了交换机，经过队列绑定交换机到达队列。

性能排序：fanout > direct >> topic。比例大约为11：10：6

一、订阅模式(Fanout Exchange)：

一个生产者，多个消费者，每一个消费者都有自己的一个队列，生产者没有将消息直接发送到队列，而是发送到了交换机，每个队列绑定交换机，生产者发送的消息经过交换机，到达队列，实现一个消息被多个消费者获取的目的。需要注意的是，如果将消息发送到一个没有队列绑定的exchange上面，那么该消息将会丢失，这是因为在rabbitMQ中exchange不具备存储消息的能力，只有队列具备存储消息的能力。

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。

1.可以理解为路由表的模式

2.这种模式不需要RouteKey

3.这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。

4.如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。

示例代码：

生产者：

[java]

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

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();

}

}

消费者1：

[java]

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, true, 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)

这种模式添加了一个路由键，生产者发布消息的时候添加路由键，消费者绑定队列到交换机时添加键值，这样就可以接收到需要接收的消息。

任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。

1.一般情况可以使用rabbitMQ自带的Exchange：”"(该Exchange的名字为空字符串，下文称其为default Exchange)。

2.这种模式下不需要将Exchange进行任何绑定(binding)操作

3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。

4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

示例代码：

生产者：

[java]

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

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

// 消息内容

String message = "删除商品， id = 1001";

channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());

System.out.println(" [x] Sent '" + message + "'");

channel.close();

connection.close();

}

}

消费者1：接收更新和删除消息

[java]

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：接收insert，update，delete的消息

[java]

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);

}

}

}

如果生产者发布了insert消息，那么消费者2可以收到，消费者 1收不到，如果发布了update或者delete消息，两个消费者都可以收到。如果发布ABC消息两个消费者都收不到，因为没有绑定这个键值。这种模式基本满足了我们的需求，但是还不够灵活，下面介绍另外一个模式。

三、通配符模式（Topic Exchange）

基本思想和路由模式是一样的，只不过路由键支持模糊匹配，符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上

1.这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。

2.这种模式需要RouteKey，也许要提前绑定Exchange与Queue。

3.在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。

4.“#”表示0个或若干个关键字，“*”表示一个关键字。如“log.*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。

5.同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。

示例代码：

生产者：

[java]

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();

}

}

消费者1：

[java]

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);

}

}

}

消费者2：

[java]

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);

}

}

}

消费者1是按需索取，并没有使用通配符模式，而是用的完全匹配，消费者2使用通配符模式，这样以item.开头的消息都会全部接收。

小结：

1.与简单模式和work模式对比，前面两种同一个消息只能被一个消费者获取，而今天的这三种模式，可以实现一个消息被多个消费者获取。

2.fanout这种模式没有加入路由器，队列与exchange绑定后，就会接收到所有的消息，其余两种增加了路由键，并且第三种增加通配符，更加便利。

本文出自https://blog.csdn.net/ww130929/article/details/72842234

转载于:https://blog.51cto.com/lookingdream/2118667

你可能感兴趣的文章

编写who命令

2.1 sikuli 中编程运行

愚公移山第一章伪代码

常见的位运算技巧总结（膜wys）

python魔法函数(二）之__getitem__、__len__、__iter__

EL表达式无法显示Model中的数据

Linux应用小技巧

ps6-工具的基础使用

关于CefSharp.WinForms的学习

灵活运用 SQL SERVER FOR XML PATH

es 加磁盘扩容

linux 参数内核

使用Azcopy在Azure上进行HBase的冷热备份还原

计组_定点数一位乘_布斯公式

linux下使用过的命令总结（未整理完）

ES6的一些文章

LeetCode 198, 213 House Robber

New Year Permutation（Floyd+并查集）

Qt编写输入法V2018超级终结版

喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！-- 愿君每日到此一游！

当前时间: 2024-12-18 19:17:15 当前IP: 18.216.177.180 联系邮箱:javaeecc@qq.com Copyright © 2020 - 2022 baihongyu.com 京ICP备2021015314号-2

强烈建议你试试无所不能的CHAT-GPT，快点击我

强烈建议你试试无所不能的CHAT-GPT，快点击我