RabbitMQ 的七种消息传递形式

一、RabbitMQ 架构简介

一图胜千言，如下：

这张图中涉及到如下一些概念：

生产者（Publisher）：发布消息到 RabbitMQ 中的交换机（Exchange）上。
交换机（Exchange）：和生产者建立连接并接收生产者的消息。
消费者（Consumer）：监听 RabbitMQ 中的 Queue 中的消息。
队列（Queue）：Exchange 将消息分发到指定的 Queue，Queue 和消费者进行交互。
路由（Routes）：交换机转发消息到队列的规则。

二、准备工作

大家知道，RabbitMQ 是 AMQP 阵营里的产品，Spring Boot 为 AMQP 提供了自动化配置依赖 spring-boot-starter-amqp，因此首先创建 Spring Boot 项目并添加该依赖，如下：

项目创建成功后，在 application.properties 中配置 RabbitMQ 的基本连接信息，如下：

spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=5672

接下来进行 RabbitMQ 配置，在 RabbitMQ 中，所有的消息生产者提交的消息都会交由 Exchange 进行再分配，Exchange 会根据不同的策略将消息分发到不同的 Queue 中。

RabbitMQ 官网介绍了如下几种消息分发的形式：



这里我主要和大家介绍前六种消息收发方式。

三、消息收发

1. Hello World

咦？这个咋没有交换机？这个其实是默认的交换机，我们需要提供一个生产者一个队列以及一个消费者。消息传播图如下：

来看看代码实现：

先来看看队列的定义：

@Configuration
public class HelloWorldConfig {public static final String HELLO_WORLD_QUEUE_NAME = "hello_world_queue";@BeanQueue queue1() {return new Queue(HELLO_WORLD_QUEUE_NAME);}
}

再来看看消息消费者的定义：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(String msg) {System.out.println("msg = " + msg);}
}

消息发送：

@SpringBootTest
class RabbitmqdemoApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testvoid contextLoads() {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello");}}

这个时候使用的其实是默认的直连交换机（DirectExchange），DirectExchange 的路由策略是将消息队列绑定到一个 DirectExchange 上，当一条消息到达 DirectExchange 时会被转发到与该条消息routing key相同的 Queue 上，例如消息队列名为 “hello-queue”，则 routingkey 为 “hello-queue” 的消息会被该消息队列接收。

2. Work queues

这种情况是这样的：

一个生产者，一个默认的交换机（DirectExchange），一个队列，两个消费者，如下图：

一个队列对应了多个消费者，默认情况下，由队列对消息进行平均分配，消息会被分到不同的消费者手中。消费者可以配置各自的并发能力，进而提高消息的消费能力，也可以配置手动 ack，来决定是否要消费某一条消息。

先来看并发能力的配置，如下：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(String msg) {System.out.println("receive = " + msg);}@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME,concurrency = "10")public void receive2(String msg) {System.out.println("receive2 = " + msg+"------->"+Thread.currentThread().getName());}
}

可以看到，第二个消费者我配置了 concurrency 为 10，此时，对于第二个消费者，将会同时存在 10 个子线程去消费消息。

启动项目，在 RabbitMQ 后台也可以看到一共有 11 个消费者。

此时，如果生产者发送 10 条消息，就会一下都被消费掉。

消息发送方式如下：

@SpringBootTest
class RabbitmqdemoApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testvoid contextLoads() {for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello");}}}

消息消费日志如下：

可以看到，消息都被第一个消费者消费了。但是小伙伴们需要注意，事情并不总是这样（多试几次就可以看到差异），消息也有可能被第一个消费者消费（只是由于第二个消费者有十个线程一起开动，所以第二个消费者消费的消息占比更大）。

当然消息消费者也可以开启手动 ack，这样可以自行决定是否消费 RabbitMQ 发来的消息，配置手动 ack 的方式如下：

spring.rabbitmq.listener.simple.acknowledge-mode=manual

消费代码如下：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)public void receive(Message message,Channel channel) throws IOException {System.out.println("receive="+message.getPayload());channel.basicAck(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)),true);}@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, concurrency = "10")public void receive2(Message message, Channel channel) throws IOException {System.out.println("receive2 = " + message.getPayload() + "------->" + Thread.currentThread().getName());channel.basicReject(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)), true);}
}

此时第二个消费者拒绝了所有消息，第一个消费者消费了所有消息。

这就是 Work queues 这种情况。

3. Publish/Subscrite

再来看发布订阅模式，这种情况是这样：

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

这种情况下，我们有四种交换机可供选择，分别是：

Direct
Fanout
Topic
Header
我分别来给大家举一个简单例子看下。

3.1. Direct

DirectExchange 的路由策略是将消息队列绑定到一个 DirectExchange 上，当一条消息到达 DirectExchange 时会被转发到与该条消息 routing key 相同的 Queue 上，例如消息队列名为 “hello-queue”，则 routingkey 为 “hello-queue” 的消息会被该消息队列接收。DirectExchange 的配置如下：

@Configuration
public class RabbitDirectConfig {public final static String DIRECTNAME = "javaboy-direct";@BeanQueue queue() {return new Queue("hello-queue");}@Bean
DirectExchange directExchange() {return new DirectExchange(DIRECTNAME, true, false);}@BeanBinding binding() {return BindingBuilder.bind(queue()).to(directExchange()).with("direct");}
}

• 首先提供一个消息队列Queue，然后创建一个DirectExchange对象，三个参数分别是名字，重启后是否依然有效以及长期未用时是否删除。
• 创建一个Binding对象将Exchange和Queue绑定在一起。
• DirectExchange和Binding两个Bean的配置可以省略掉，即如果使用DirectExchange，可以只配置一个Queue的实例即可。

再来看看消费者：

@Component
public class DirectReceiver {@RabbitListener(queues = "hello-queue")public void handler1(String msg) {System.out.println("DirectReceiver:" + msg);}
}

通过 @RabbitListener 注解指定一个方法是一个消息消费方法，方法参数就是所接收到的消息。然后在单元测试类中注入一个 RabbitTemplate 对象来进行消息发送，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void directTest() {rabbitTemplate.convertAndSend("hello-queue", "hello direct!");}
}

最终执行结果如下：

3.2. Fanout

FanoutExchange 的数据交换策略是把所有到达 FanoutExchange 的消息转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用，FanoutExchange 配置方式如下：

@Configuration
public class RabbitFanoutConfig {public final static String FANOUTNAME = "sang-fanout";@BeanFanoutExchange fanoutExchange() {return new FanoutExchange(FANOUTNAME, true, false);}@BeanQueue queueOne() {return new Queue("queue-one");}@BeanQueue queueTwo() {return new Queue("queue-two");}@BeanBinding bindingOne() {return BindingBuilder.bind(queueOne()).to(fanoutExchange());}@BeanBinding bindingTwo() {return BindingBuilder.bind(queueTwo()).to(fanoutExchange());}
}

在这里首先创建 FanoutExchange，参数含义与创建 DirectExchange 参数含义一致，然后创建两个 Queue，再将这两个 Queue 都绑定到 FanoutExchange 上。接下来创建两个消费者，如下：

@Component
public class FanoutReceiver {@RabbitListener(queues = "queue-one")public void handler1(String message) {System.out.println("FanoutReceiver:handler1:" + message);}@RabbitListener(queues = "queue-two")public void handler2(String message) {System.out.println("FanoutReceiver:handler2:" + message);}
}

两个消费者分别消费两个消息队列中的消息，然后在单元测试中发送消息，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void fanoutTest() {rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME, null, "hello fanout!");}
}

注意这里发送消息时不需要 routingkey，指定 exchange 即可，routingkey 可以直接传一个 null。

最终执行日志如下：

3.3. Topic

TopicExchange 是比较复杂但是也比较灵活的一种路由策略，在 TopicExchange 中，Queue 通过 routingkey 绑定到 TopicExchange 上，当消息到达 TopicExchange 后，TopicExchange 根据消息的 routingkey 将消息路由到一个或者多个 Queue 上。TopicExchange 配置如下：

@Configuration
public class RabbitTopicConfig {public final static String TOPICNAME = "sang-topic";@BeanTopicExchange topicExchange() {return new TopicExchange(TOPICNAME, true, false);}@BeanQueue xiaomi() {return new Queue("xiaomi");}@BeanQueue huawei() {return new Queue("huawei");}@BeanQueue phone() {return new Queue("phone");}@BeanBinding xiaomiBinding() {return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");}@BeanBinding huaweiBinding() {return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");}@BeanBinding phoneBinding() {return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");}
}

• 首先创建 TopicExchange，参数和前面的一致。然后创建三个 Queue，第一个 Queue 用来存储和 “xiaomi” 有关的消息，第二个 Queue 用来存储和 “huawei” 有关的消息，第三个 Queue 用来存储和 “phone” 有关的消息。
• 将三个 Queue 分别绑定到 TopicExchange 上，第一个 Binding 中的 “xiaomi.#” 表示消息的 routingkey 凡是以 “xiaomi” 开头的，都将被路由到名称为 “xiaomi” 的 Queue 上，第二个 Binding 中的 “huawei.#” 表示消息的 routingkey 凡是以 “huawei” 开头的，都将被路由到名称为 “huawei” 的 Queue 上，第三个 Binding 中的 “#.phone.#” 则表示消息的 routingkey 中凡是包含 “phone” 的，都将被路由到名称为 “phone” 的 Queue 上。

接下来针对三个 Queue 创建三个消费者，如下：

@Component
public class TopicReceiver {@RabbitListener(queues = "phone")public void handler1(String message) {System.out.println("PhoneReceiver:" + message);}@RabbitListener(queues = "xiaomi")public void handler2(String message) {System.out.println("XiaoMiReceiver:"+message);}@RabbitListener(queues = "huawei")public void handler3(String message) {System.out.println("HuaWeiReceiver:"+message);}
}

然后在单元测试中进行消息的发送，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void topicTest() {rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"xiaomi.news","小米新闻..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"huawei.news","华为新闻..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"xiaomi.phone","小米手机..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"huawei.phone","华为手机..");rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"phone.news","手机新闻..");}
}

根据 RabbitTopicConfig 中的配置，第一条消息将被路由到名称为 “xiaomi” 的 Queue 上，第二条消息将被路由到名为 “huawei” 的 Queue 上，第三条消息将被路由到名为 “xiaomi” 以及名为 “phone” 的 Queue 上，第四条消息将被路由到名为 “huawei” 以及名为 “phone” 的 Queue 上，最后一条消息则将被路由到名为 “phone” 的 Queue 上。

3.4. Header

HeadersExchange 是一种使用较少的路由策略，HeadersExchange 会根据消息的 Header 将消息路由到不同的 Queue 上，这种策略也和 routingkey无关，配置如下：

@Configuration
public class RabbitHeaderConfig {public final static String HEADERNAME = "javaboy-header";@BeanHeadersExchange headersExchange() {return new HeadersExchange(HEADERNAME, true, false);}@BeanQueue queueName() {return new Queue("name-queue");}@BeanQueue queueAge() {return new Queue("age-queue");}@BeanBinding bindingName() {Map<String, Object> map = new HashMap<>();map.put("name", "sang");return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();}@BeanBinding bindingAge() {return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();}
}

这里的配置大部分和前面介绍的一样，差别主要体现的 Binding 的配置上，第一个 bindingName 方法中，whereAny 表示消息的 Header 中只要有一个 Header 匹配上 map 中的 key/value，就把该消息路由到名为 “name-queue” 的 Queue 上，这里也可以使用 whereAll 方法，表示消息的所有 Header 都要匹配。whereAny 和 whereAll 实际上对应了一个名为 x-match 的属性。bindingAge 中的配置则表示只要消息的 Header 中包含 age，不管 age 的值是多少，都将消息路由到名为 “age-queue” 的 Queue 上。

接下来创建两个消息消费者：

@Component
public class HeaderReceiver {@RabbitListener(queues = "name-queue")public void handler1(byte[] msg) {System.out.println("HeaderReceiver:name:"+ new String(msg, 0, msg.length));}@RabbitListener(queues = "age-queue")public void handler2(byte[] msg) {System.out.println("HeaderReceiver:age:"+ new String(msg, 0, msg.length));}
}

注意这里的参数用 byte 数组接收。然后在单元测试中创建消息的发送方法，这里消息的发送也和 routingkey 无关，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void headerTest() {Message nameMsg = MessageBuilder.withBody("hello header! name-queue".getBytes()).setHeader("name", "sang").build();Message ageMsg = MessageBuilder.withBody("hello header! age-queue".getBytes()).setHeader("age", "99").build();rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);}
}

这里创建两条消息，两条消息具有不同的 header，不同 header 的消息将被发到不同的 Queue 中去。

最终执行效果如下：

4. Routing

这种情况是这样：

一个生产者，一个交换机，两个队列，两个消费者，生产者在创建 Exchange 后，根据 RoutingKey 去绑定相应的队列，并且在发送消息时，指定消息的具体 RoutingKey 即可。

如下图：

这个就是按照 routing key 去路由消息，我这里就不再举例子了，大家可以参考 3.3.1 小结。

5. Topics

这种情况是这样：

一个生产者，一个交换机，两个队列，两个消费者，生产者创建 Topic 的 Exchange 并且绑定到队列中，这次绑定可以通过 * 和 # 关键字，对指定 RoutingKey 内容，编写时注意格式 xxx.xxx.xxx 去编写。

如下图：

这个我也就不举例啦，前面 3.3.3 小节已经举过例子了，不再赘述。