1.Exchange的概念
2.Exchange的绑定
3.Exchange的Fanout
4.Exchange的direct
5.Exchange的Topics
1.Exchange的概念
在我们之前练习mq的基本使用的时候,我们发送一个信息,看起来是直接把消息发送给了队列,其实,Rabbitmq消息传递模型的核心思想是:生产者生产的消息不会发送给队列,而是发送给交换机。交换机工作的内容也十分简单,一方面接受来自生产者的消息,另一方面将他们推入队列。交换机知道如何准确地处理消息,到底是发送给一个或者多个队列,还是丢掉,都由交换机决定。

所以看看之前我们写的代码,我们是把消息发送给了一个无名的交换机
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
第一个参数就是交换机的名称,空字符串表示无名交换机,消息能路由到哪个队列中是由routingKey(bindingkey)绑定 key 指定的。
2.Exchange的绑定
因为交换机知道要把消息发送给哪个队列,但是我们需要将队列和路由器进行绑定。

绑定的方式有Fanout(广播),Direct exchange(全匹配),Topics(主题匹配)。
3.Exchange的FanoutFanout类型的exchange简单粗暴,就是广播,将受到的所有消息广播到它知道的所有队列中。

消费者1:
public class Consumer1 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, "fanout");/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("消费者1等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者1控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
消费者2:
public class Consumer2 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, "fanout");/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("消费者2等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者2控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
生产者:
public class Producer {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception {try (Channel channel = RabbitMqUtils.getChannel()) {/** * 声明一个 exchange * 1.exchange 的名称 * 2.exchange 的类型 */ channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); Scanner sc = new Scanner(System.in); System.out.println("请输入信息");while (sc.hasNext()) { String message = sc.nextLine(); channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println("生产者发出消息" + message); } } }}
我们连续发送十条消息:消费者1和消费者2同时受到了消息:

成功进行轮询分发。
4.Exchange的direct
我们在上面演示的是把消息广播给全部消费者,但是我们希望做一些改变,比如只发给一些特定的队列,有一些队列不发。Fanout这种方式对我们来说灵活性不是很高。
所以我们采取direct的方式,绑定key和队列一一对应,这样我们就可以根据绑定key来选择要发送的队列。

但是如果k1和k2相同,就有点像fanout的感觉了,也到达了轮询分发的效果。
消费者1代码:
public class Consumer1 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,"k1"); System.out.println("消费者1等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者1控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
消费者2代码:
public class Consumer2 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,"k2"); System.out.println("消费者2等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者2控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
生产者代码:
public class Producer {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception {try (Channel channel = RabbitMqUtils.getChannel()) {/** * 声明一个 exchange * 1.exchange 的名称 * 2.exchange 的类型 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); Scanner sc = new Scanner(System.in); System.out.println("请输入信息");while (sc.hasNext()) { String message = sc.nextLine(); channel.basicPublish(EXCHANGE_NAME, "k1", null, message.getBytes("UTF-8")); System.out.println("生产者发出消息" + message); } } }}
生产者发送消息:
消费者1有消息,消费者2无消息:

5.Exchange的Topics
之前我们从广播分发的fanout交换机,到了可选择的direct交换机,从而实现有选择性地对消息进行发送和消费。
但是,新的需求又产生了,假设现在有一个日志系统,我们对多台服务器用队列进行发送日志,进行保存。
假设现在有了一个子分类,info.feign , info.service,我们只想要保存info.service的消息,不想保存info.feign的消息,我们就必须要使用topic模式。
topic交换机的消息绑定key不能随意编写,它必须是一个单词列表,用.分开。这些单词是任意单词,比如"info.user , info.store"等等。
在这个规则中,还有两条规则:“*”(星号)可以代替一个单词“#”(井号)可以代替多个单词
例如:Q1绑定的是三个单词的字符串 (.orange.)
Q2绑定的是最后一个单词是rabbit的字符串 (..rabbit)
第二个绑定关系是lazy后有任意个单词的字符串(lazy.#)

所以实际会出现下面这种情况
绑定key | 会收到的队列 |
---|---|
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 | 无队列接收 |
quick.orange.male.rabbit | 无队列接收 |
lazy.orange.male.rabbit | Q2 |
消费者1:
public class Consumer1 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,"*.orange.*"); System.out.println("消费者1等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者1控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
消费者2:
public class Consumer2 {private final static String EXCHANGE_NAME = "test"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel();//声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);/** * 生成一个临时的队列 队列的名称是随机的 * 当消费者断开和该队列的连接时 队列自动删除 */ String queueName = channel.queueDeclare().getQueue();//交换机和队列绑定,因为是fanout类型,所以无需路由键 channel.queueBind(queueName,EXCHANGE_NAME,"lazy.#"); System.out.println("消费者2等待接收消息.........");//消息如何进行消费的业务逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody()); System.out.println("消费者2控制台接收并打印消息:"+message); };//取消消费的一个回调接口 如在消费的时候队列被删除掉了 CancelCallback cancelCallback = (consumerTag) -> { System.out.println("消息消费被中断"); }; channel.basicConsume(queueName, true, deliverCallback, cancelCallback); }}
生产者:
publicclassProducer {private final static String EXCHANGE_NAME = "test";publicstaticvoidmain(String[] args) throws Exception {try (Channel channel = RabbitMqUtils.getChannel()) {/** * 声明一个 exchange * 1.exchange 的名称 * 2.exchange 的类型 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); Scanner sc = new Scanner(System.in); System.out.println("请输入信息");while (sc.hasNext()) { String message = sc.nextLine(); String routingKey = null;//当发送的消息不同,路右键也不同if(message.equals("1")){ routingKey = "1.orange.1"; }elseif(message.equals("2")){ routingKey = "lazy.1.1"; }elseif(message.equals("3")){ routingKey = "lazy.orange.1"; }elseif(message.equals("4")){ routingKey = "orange.1"; } channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8")); System.out.println("生产者发出消息" + message); } } }}
演示效果:
按照我们的期望:
消息1和2都会被各自的队列收到
消息3会被两个队列收到
消息4没有队列收到




版权声明:本文内容来自个人博客segmentfault:苏凌峰,遵循CC 4.0 BY-SA版权协议上原文接及本声明。
本作品采用知识共享署名-非商业性使用-禁止演绎 2.5 中国大陆许可协议进行可。
原文链接:https://segmentfault.com/a/1190000041993717
如有涉及到侵权,请联系,将立即予以删除处理。
在此特别鸣谢原作者的创作。
此篇文章的所有版权归原作者所有,与本公众号无关,商业转载建议请联系原作者,非商业转载请注明出处。