图文实践 RabbitMQ 不同类型交换机消息投递机制(Java 版)

    交换机有四种类型,每种类型有不同的路由策略,这块也是入门的难点,在初学的过程中对消息的路由机制也是一直不是很理解,本文通过实践总结对不同类型交换机的路由策略分别进行了讲解。

    Exchange 参数介绍

    • Name:交换机名称
    • Type:交换机类型 direct、topic、fanout、headers
    • Durability:是否需要持久化
    • Auto Delete:最后一个绑定到 Exchange 上的队列删除之后自动删除该 Exchange
    • Internal:当前 Exchange 是否应用于 RabbitMQ 内部使用,默认false。
    • Arguments:扩展参数

    Exchange 四种类型

    • direct:不需要 Exchange 进行绑定,根据 RoutingKey 匹配消息路由到指定的队列。
    • topic:生产者指定 RoutingKey 消息根据消费端指定的队列通过模糊匹配的方式进行相应转发,两种通配符模式:
      • :可匹配一个或多个关键字
      • *:只能匹配一个关键字
    • fanout:这种模式只需要将队列绑定到交换机上即可,是不需要设置路由键的。
    • headers:根据发送消息内容中的 headers 属性来匹配

    添加maven依赖

    Spring Boot 项目的 pom.xml 文件中引入 amqp-client 启动器

    direct 通过 RoutingKey 匹配消息路由到指定的队列,因此也可以无需指定交换机,在不指定交换机的情况下会使用 AMQP default 默认的交换机,另外在消息投递时要注意 RoutingKey 要完全匹配才能被队列所接收,否则消息会被丢弃的。

    上图三个队列,第一个队列的 Binding routingKey 为 black,第二个队列和第三个队列的 Binding routingKey 为 green 和 green,也很清晰的能看到消息投递 1 仅被 Queue1 接收,而 消息投递 2 同时可以被广播到 Queue2 和 Queue3,这是因为 Queue2 和 Queue3 的路由键是相同的,再一次的说明了交换机的 direct 模式是通过 RoutingKey 进行消息路由的

    1. package com.may.rabbitmq.direct.helloworld;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import com.rabbitmq.client.ConnectionFactory;
    5. public class Producer {
    6. public static void main(String[] args) throws Exception {
    7. // 创建链接工厂
    8. ConnectionFactory connectionFactory = new ConnectionFactory();
    9. connectionFactory.setHost("127.0.0.1");
    10. connectionFactory.setPort(5672);
    11. connectionFactory.setVirtualHost("/");
    12. // 通过链接工厂创建链接
    13. Connection connection = connectionFactory.newConnection();
    14. // 通过链接创建通道(channel)
    15. Channel channel = connection.createChannel();
    16. // 通过 channel 发送数据
    17. // exchange:交换机,如果不传默认为 AMQP default
    18. String directExchangeName = "direct_exchange_name";
    19. String directRoutingKey = "direct_routingKey";
    20. String directMsg = "direct hello world";
    21. channel.basicPublish(directExchangeName, directRoutingKey, null, directMsg.getBytes());
    22. // 关闭链接
    23. channel.close();
    24. connection.close();
    25. }

    构建消费者

    交换机类型之 topic

    生产者指定 RoutingKey ,消费端根据指定的队列通过模糊匹配的方式进行相应转发,两种通配符模式如下:

    • #:可匹配一个或多个关键字
    • *:只能匹配一个关键字

    图文实践 RabbitMQ 不同类型交换机消息投递机制(Java 版) - 图1

    上图展示了交换机 Topic 模式的消息流转过程,Queue1 的路由键通过使用 * 符合匹配到了 black.test1 和 black.test2 但是 black.test3.1 由于有多个关键字是匹配不到的。另一个队列 Queue2 使用了 # 符号即可以一个也可以匹配多个关键字,同时匹配到了 black.test4 和 black.test5.1。

    构建生产者

    1. package com.may.rabbitmq.topic.helloworld;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. public class Producer {
    5. public static void main(String[] args) throws Exception {
    6. // 创建链接工厂
    7. ConnectionFactory connectionFactory = new ConnectionFactory();
    8. connectionFactory.setHost("127.0.0.1");
    9. connectionFactory.setPort(5672);
    10. connectionFactory.setVirtualHost("/");
    11. // 通过链接工厂创建链接
    12. Connection connection = connectionFactory.newConnection();
    13. // 通过链接创建通道(channel)
    14. Channel channel = connection.createChannel();
    15. // 通过 channel 发送数据
    16. // exchange:交换机,如果不传默认为 AMQP default
    17. String topicExchangeName = "topic_exchange_name";
    18. String topicRoutingKey1 = "topic_routingKey.test1";
    19. String topicRoutingKey2 = "topic_routingKey.test2";
    20. String topicRoutingKey3 = "topic_routingKey.test3.test03";
    21. String topicMsg = "topic hello world";
    22. channel.basicPublish(topicExchangeName, topicRoutingKey1, null, topicMsg.getBytes());
    23. channel.basicPublish(topicExchangeName, topicRoutingKey2, null, topicMsg.getBytes());
    24. channel.basicPublish(topicExchangeName, topicRoutingKey3, null, topicMsg.getBytes());
    25. // 关闭链接
    26. channel.close();
    27. connection.close();
    28. }
    29. }

    构建消费者

    由于和 direct 模式是相同的,仅列出改动部分

    1. https://github.com/Q-Angelo/project-training/tree/master/rabbitmq/helloworld-topic

    与 direct 和 topic 两种类型不同的是这种模式只需要将队列绑定到交换机上即可,是不需要设置路由键的,便可将消息转发到绑定的队列上,正式由于不需要路由键,所以 fanout 也是四个交换机类型中最快的一个,如果是做广播模式的就很适合。

    下图展示了 fanout 类型交换机的消息流转过程

    构建生产者

    只需要设置交换机类型为 fanout 即可,路由键无为空,设置了也是不会生效的

    构建消费者

    1. package com.may.rabbitmq.fanout.helloworld;
    2. import com.rabbitmq.client.*;
    3. import java.io.IOException;
    4. import java.util.concurrent.TimeUnit;
    5. public class Consumer {
    6. public static void main(String[] args) throws Exception {
    7. ConnectionFactory connectionFactory = new ConnectionFactory();
    8. connectionFactory.setHost("127.0.0.1");
    9. connectionFactory.setPort(5672);
    10. // 通过链接工厂创建链接
    11. Connection connection = connectionFactory.newConnection();
    12. // 通过链接创建通道(channel)
    13. Channel channel = connection.createChannel();
    14. // 定义信息
    15. String fanoutExchangeName = "fanout_exchange_name";
    16. String exchangeType = "fanout";
    17. String fanoutRoutingKey = "";
    18. String fanoutQueueName = "fanout_queue";
    19. // 声明一个交换机
    20. channel.exchangeDeclare(fanoutExchangeName, exchangeType, true, false, null);
    21. // 声明一个队列
    22. channel.queueDeclare(fanoutQueueName, true, false, false, null);
    23. // 绑定关系(队列、交换机、路由键)
    24. channel.queueBind(fanoutQueueName, fanoutExchangeName, fanoutRoutingKey);
    25. // 创建消费者
    26. // springboot 从 1.5.9 升级到 2.0.0,QueueingConsumer 报错(Cannot resolve symbol 'QueueingConsumer')没有这个类,改为使用 DefaultConsumer
    27. DefaultConsumer consumer = new DefaultConsumer(channel) {
    28. @Override
    29. public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    30. super.handleDelivery(consumerTag, envelope, properties, body);
    31. String message = new String(body, "UTF-8");
    32. System.out.printf("in consumer B (delivery tag is %d): %s\n", envelope.getDeliveryTag(), message);
    33. try {
    34. TimeUnit.MILLISECONDS.sleep(200);
    35. } catch (InterruptedException e) {
    36. }
    37. channel.basicAck(envelope.getDeliveryTag(), false);
    38. }
    39. };
    40. // 6. 设置 channel
    41. channel.basicConsume(fanoutQueueName, false, consumer);
    42. System.out.println("消费端启动成功!");
    43. }

    交换机类型之 headers

    该类型的交换机是根据发送消息内容中的 headers 属性来匹配的,headers 类型的交换机基本上不会用到,因此这里也不会过多介绍,掌握以上三种类型的交换机模型在平常的业务场景中就足够了。