RabbitMQ 中的六大工作模式介绍与使用

文章目录

- 简单队列（Simple Queue）模式
- - 配置类定义
  - 消费者定义
  - 发送消息测试消费
- 工作队列（Work Queues）模式
- - 配置类定义
  - 消费者定义
  - 发送消息测试消费
  - 负载均衡调优
- 发布/订阅（Publish/Subscribe）模式
- - 配置类定义
  - 消费者定义
  - 发送消息测试消费
- 路由（Routing）模式
- - 配置类定义
  - 消费者定义
  - 发送消费测试消费
- 主题（Topics）模式
- - 配置类定义
  - 消费者定义
  - 发送消息测试消费
- RPC（Remote Procedure Call）模式
- - 配置类定义
  - 请求与响应实体类
  - 消费者定义
  - 发送消息测试消费

简单队列（Simple Queue）模式

结构：一个生产者对应一个消费者，生产者将消息发送到指定队列，消费者从该队列获取消息。
工作流程：生产者创建一个消息并发送到 RabbitMQ 的队列中，消费者从这个队列里取出消息进行处理。
应用场景：适用于任务处理逻辑简单，对消息处理速度要求不高，且消息处理顺序要求不严格的场景，比如简单的日志收集系统。

以下是简单队列（Simple Queue）模式的使用案例。

配置类定义

package test;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class SimpleQueueConfig {@Beanpublic DirectExchange simpleQueueExchange() {return ExchangeBuilder// 指定名称为simple-queue-exchange的交换机.directExchange("simple-queue-exchange")// 声明为持久化交换机.durable(true).build();}@Beanpublic Queue simpleQueue() {return QueueBuilder// 指定一个名称为simple-queue的持久化队列.durable("simple-queue").build();}@Beanpublic Binding simpleQueueToSimpleQueueExchangebinding() {// 绑定队列给交换机，根据队列名为路由键return BindingBuilder.bind(simpleQueue()).to(simpleQueueExchange()).withQueueName();}}

消费者定义

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@Slf4j
public class SimpleQueueConsumer {@RabbitListener(queues = "simple-queue")public void consumer(String message) {log.info("消费者接收消息：{}", message);}}

发送消息测试消费

package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend("simple-queue", "hello world!");}}

控制台输出：

在这里插入图片描述

工作队列（Work Queues）模式

结构：一个生产者对应多个消费者，多个消费者共同从一个队列中获取消息进行处理。
工作流程：生产者将消息发送到队列，多个消费者从该队列中获取消息（默认采用轮询的负载均衡策略），一条消息只会被一个消费者处理。
应用场景：适用于需要将大量任务分发给多个工作者并行处理的场景，例如订单处理、图片处理等。

以下是工作队列（Work Queues）模式的使用案例。

配置类定义

定义一个 DirectExchange 类型的交换机与两个队列，通过队列名作为 RoutingKey 进行绑定。

package test;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class WorkQueueConfig {@Beanpublic DirectExchange workQueueExchange(){return ExchangeBuilder.directExchange("work-queue-exchange").durable(true).build();}@Beanpublic Queue workQueue(){return QueueBuilder.durable("work-queue").build();}@Beanpublic Binding workQueueToWorkQueueExchangeBinding(){return BindingBuilder.bind(workQueue()).to(workQueueExchange()).withQueueName();}}

消费者定义

定义两个消费者，分别监听两个不同的队列。

消费者 1：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class WorkQueueConsumer01 {@RabbitListener(queues = "work-queue")public void receive(String message){log.info("WorkQueueConsumer01 receive a message:{}",message);}}

消费者 2：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class WorkQueueConsumer02 {@RabbitListener(queues = "work-queue")public void receive(String message){log.info("WorkQueueConsumer02 receive a message:{}",message);}}

发送消息测试消费

向 work-queue-exchange 交换机中发送 10 条消息。

package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend("work-queue-exchange", "work-queue", "hello world!" + i);}}}

控制台输出：

在这里插入图片描述

在上例结果，消费者 1 与消费者 2 对消息进行了轮询处理。

负载均衡调优

在工作队列模式下，多个消费者处理消息默认按照轮询的规则进行处理。但是如果其中某个消费者的处理能力比其他消费者更为强大，那么它实际上可以处理更多的消息，如果此时还是按照轮询规则来处理消息，这样是对该消费者能力的一种浪费。

对于上述情况，在 RabbitMQ 中支持调整消费者处理消息时的负载均衡策略，具体而言，即通过设置预取数量（prefetch count）参数，以实现更灵活的消息分发策略，即让处理能力更强大的消费者处理更多的消息。

在 Spring Boot 中，可以通过设置 spring.rabbitmq.listener.simple.prefetch 参数来控制消费者的预取数量：

spring:rabbitmq:listener:simple:prefetch: 1

将预取数量设置为 1，则每个消费者在处理完当前消息并确认（ACK）之后才会接收下一条消息，从而避免了某个消费者负载过重的情况。

模拟当某个消费者处理能力较弱：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class WorkQueueConsumer01 {@RabbitListener(queues = "work-queue")public void receive(String message) throws InterruptedException {// 模拟WorkQueueConsumer01处理能力较弱Thread.sleep(1);log.info("WorkQueueConsumer01 receive a message:{}",message);}}

再次向交换机发送 10 条消息，控制台输出：

在这里插入图片描述

在上述结果中，由于 WorkQueueConsumer01 处理消息的能力更弱，所以更多的消息被分摊到了 WorkQueueConsumer02 处理。

发布/订阅（Publish/Subscribe）模式

结构：包含一个生产者、一个交换机（Exchange）和多个队列，多个消费者分别监听不同的队列。
工作流程：生产者将消息发送到交换机，交换机将消息广播到绑定到它的所有队列，每个绑定的队列对应的消费者都能收到消息。
应用场景：适用于一个消息需要被多个不同的服务或模块接收和处理的场景，如系统的通知功能。

以下是发布/订阅（Publish/Subscribe）模式的使用案例。

配置类定义

定义一个 FanoutExchange 类型的交换机与三个队列，让三个队列绑定到该交换机。

FanoutExchange 交换机不需要指定 RoutingKey。因为 FanoutExchange 交换机会将消息统一发送给与其绑定的所有队列，指定 RoutingKey 实际上没有任何意义。

package test;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class PublishSubscribeConfig {@Beanpublic Queue publishSubscribeQueue1() {return new Queue("publish-subscribe-queue-1");}@Beanpublic Queue publishSubscribeQueue2() {return new Queue("publish-subscribe-queue-2");}@Beanpublic Queue publishSubscribeQueue3() {return new Queue("publish-subscribe-queue-3");}@Beanpublic FanoutExchange publishSubscribeFanoutExchange() {return new FanoutExchange("publish-subscribe-fanout-exchange");}@Beanpublic Binding publishSubscribeBinding1() {return BindingBuilder.bind(publishSubscribeQueue1()).to(publishSubscribeFanoutExchange());}@Beanpublic Binding publishSubscribeBinding2() {return BindingBuilder.bind(publishSubscribeQueue2()).to(publishSubscribeFanoutExchange());}@Beanpublic Binding publishSubscribeBinding3() {return BindingBuilder.bind(publishSubscribeQueue3()).to(publishSubscribeFanoutExchange());}}

消费者定义

定义三个消费者，分别监听三个不同的队列。

第一个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class PublishSubscribeConsumer01 {@RabbitListener(queues = "publish-subscribe-queue-1")public void receive(String message) {log.info("PublishSubscribeConsumer 01 receive a Message:{}",message);}}

第二个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class PublishSubscribeConsumer02 {@RabbitListener(queues = "publish-subscribe-queue-2")public void receive(String message) {log.info("PublishSubscribeConsumer 02 receive a message:{}",message);}}

第三个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class PublishSubscribeConsumer03 {@RabbitListener(queues = "publish-subscribe-queue-3")public void receive(String message) {log.info("PublishSubscribeConsumer 03 receive a message:{}",message);}}

发送消息测试消费

向 publish-subscribe-fanout-exchange 发送一条消息。

package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend("publish-subscribe-fanout-exchange","","hello world!");}}

控制台输出：

在这里插入图片描述

路由（Routing）模式

结构：同样包含生产者、交换机和多个队列，但交换机类型为直连交换机（Direct Exchange），队列通过路由键（routing key）绑定到交换机。
工作流程：生产者发送消息时指定路由键，交换机根据路由键将消息路由到与之绑定的队列，只有绑定了对应路由键的队列才能收到消息。
应用场景：适用于根据不同的业务规则将消息路由到不同队列的场景，如根据日志的级别（info、error 等）将日志消息路由到不同的队列。

以下是路由（Routing）模式的使用案例。

配置类定义

定义一个 DirectExchange，因为 DirectExchange 可以通过 RoutingKey 精确匹配消息，当然也可以使用其他类型的交换机，如 ToppicExchange。

定义三个不同名称的队列，其中两个队列绑定相同的 RoutingKey，而另外一个单独绑定一个队列。

package test;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RoutingConfig {@Beanpublic DirectExchange routingExchange() {return ExchangeBuilder.directExchange("routing-exchange").durable(true).build();}@Beanpublic Queue routingQueue1() {return QueueBuilder.durable("routing-queue-1").build();}@Beanpublic Queue routingQueue2() {return QueueBuilder.durable("routing-queue-2").build();}@Beanpublic Queue routingQueue3() {return QueueBuilder.durable("routing-queue-3").build();}@Beanpublic Binding routingQueue1ToRoutingExchangeBinding() {return BindingBuilder.bind(routingQueue1()).to(routingExchange()).with("routing-key-1");}@Beanpublic Binding routingQueue2ToRoutingExchangeBinding() {return BindingBuilder.bind(routingQueue2()).to(routingExchange()).with("routing-key-2");}@Beanpublic Binding routingQueue3ToRoutingExchangeBinding() {return BindingBuilder.bind(routingQueue3()).to(routingExchange()).with("routing-key-2");}}

消费者定义

定义三个消费者，分别监听三个不同的队列。

第一个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class RoutingQueueConsumer01 {@RabbitListener(queues = "routing-queue-1")public void receive(String message){log.info("RoutingQueueConsumer01 receive a message:{}",message);}}

第二个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class RoutingQueueConsumer02 {@RabbitListener(queues = "routing-queue-2")public void receive(String message){log.info("RoutingQueueConsumer02 receive a message:{}",message);}}

第三个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class RoutingQueueConsumer03 {@RabbitListener(queues = "routing-queue-3")public void receive(String message){log.info("RoutingQueueConsumer03 receive a message:{}",message);}}

发送消费测试消费

向交换机发送消息，并指定 RoutingKey 为其中的两个。

package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend("routing-exchange", "routing-key-1", "hello world by routing-key-1!");rabbitTemplate.convertAndSend("routing-exchange", "routing-key-2", "hello world by routing-key-2!");}}

控制台输出：

在这里插入图片描述

上述结果中，三个消费者都接受到了消息并成功处理。但实际上发送消息时，交换机只向两个 RoutingKey 发送消息，而另外一个 RoutingKey 并没有发送。这样的情况下，三个消息者监听三个不同队列仍然能够接受消息并处理，这是因为它们监听的队列所绑定的 RoutingKey 都被交换机投递了消息。

主题（Topics）模式

结构：包含生产者、主题交换机（Topic Exchange）和多个队列，队列通过绑定键（binding key）绑定到交换机，绑定键可以使用通配符。
工作流程：生产者发送消息时指定路由键，交换机根据绑定键和路由键的匹配规则将消息路由到相应的队列。
应用场景：适用于需要根据消息的主题进行灵活路由的场景，如新闻分类订阅，用户可以根据不同的主题订阅不同的新闻。

生产者发送包含主题的消息，主题由一个或多个单词组成，单词之间使用点号 . 进行分隔。消费者可以使用通配符 * 和 # 对主题进行模糊匹配：

*：匹配一个单词，可以代表任意一个词。
#：匹配零个或多个单词，可以代表一个或多个词。

以下是主题（Topics）模式的使用案例。

配置类定义

定义一个 TopicExchange，分别通过 log.info.* 与 log.error.# 的 RoutingKey 绑定到两个不同的队列。

package test;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class TopicConfig {@Beanpublic Queue topicQueue1() {return new Queue("topic-queue-1");}@Beanpublic Queue topicQueue2() {return new Queue("topic-queue-2");}@Beanpublic TopicExchange topicExchange() {return new TopicExchange("topic-exchange");}@Beanpublic Binding bindingExchange() {return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("log.info.*");}@Beanpublic Binding bindingExchange2() {return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("log.error.#");}}

消费者定义

定义两个消费者分别监听两个不同队列。

第一个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class TopicQueueConsumer01 {@RabbitListener(queues = "topic-queue-1")public void receive(String message){log.info("TopicQueueConsumer01 receive a message:{}",message);}}

第二个消费者：

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class TopicQueueConsumer02 {@RabbitListener(queues = "topic-queue-2")public void receive(String message){log.info("TopicQueueConsumer02 receive a message:{}",message);}}

发送消息测试消费

向 topic-exchange 交换机发送六条消息，通过不同的 RoutingKey。

package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend("topic-exchange", "log.info.base", "log.info.base message!");rabbitTemplate.convertAndSend("topic-exchange", "log.info.customer", "log.info.customer message!");rabbitTemplate.convertAndSend("topic-exchange", "log.info.shop.goods", "log.info.shop.goods message!");rabbitTemplate.convertAndSend("topic-exchange", "log.error.base", "log.error.base message!");rabbitTemplate.convertAndSend("topic-exchange", "log.error.customer", "log.error.customer message!");rabbitTemplate.convertAndSend("topic-exchange", "log.error.shop.goods", "log.error.shop.goods message!");}}

控制台输出：

在这里插入图片描述

RPC（Remote Procedure Call）模式

结构：包含客户端（生产者）、服务端（消费者）、请求队列和响应队列。
工作流程：客户端发送请求消息到请求队列，并设置一个回调队列用于接收响应；服务端从请求队列获取请求消息，处理后将响应消息发送到回调队列，客户端从回调队列获取响应。
应用场景：适用于需要在分布式系统中进行远程过程调用的场景，如微服务之间的调用。

以下是主题（Topics）模式的使用案例。

配置类定义

定义一个交换机与一个队列，通过队列名进行绑定。

package test;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RpcConfig {@Beanpublic Queue rpcQueue() {return new Queue("rpc-queue", true);}@Beanpublic DirectExchange rpcExchange() {return new DirectExchange("rpc-exchange");}@Beanpublic Binding rpcQueueToRpcExchangeBinding() {return BindingBuilder.bind(rpcQueue()).to(rpcExchange()).withQueueName();}@Beanpublic Jackson2JsonMessageConverter jackson2JsonMessageConverter() {return new Jackson2JsonMessageConverter();}}

请求与响应实体类

创建 RPC 请求与响应实体类。

package test;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;@Data
@AllArgsConstructor
@NoArgsConstructor
public class RpcRequest implements Serializable {private String message;}

package test;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;@Data
@AllArgsConstructor
@NoArgsConstructor
public class RpcResponse implements Serializable {private int result;}

消费者定义

定义一个消费者从队列中接收消息。

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
@Slf4j
public class RpcServer {@RabbitListener(queues = "rpc-queue")public RpcResponse handleRpcRequest(RpcRequest request) {log.info("RPC接收的消息：{}", request);return new RpcResponse(0);}}

发送消息测试消费

测试 RPC 消息的发送。

package test;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;@SpringBootApplication
@Slf4j
public class SpringBootApp {public static void main(String[] args) {ApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);String exchange = "rpc-exchange";String routingKey = "rpc-queue";RpcRequest rpcRequest = new RpcRequest("Rpc message!");RpcResponse rpcResponse = (RpcResponse) rabbitTemplate.convertSendAndReceive(exchange, routingKey, rpcRequest);log.info("RPC返回的结果：{}", rpcResponse);}}