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 {

    @Bean
    public DirectExchange simpleQueueExchange() {
        return ExchangeBuilder
                // 指定名称为simple-queue-exchange的交换机
                .directExchange("simple-queue-exchange")
                // 声明为持久化交换机
                .durable(true)
                .build();
    }

    @Bean
    public Queue simpleQueue() {
        return QueueBuilder
                // 指定一个名称为simple-queue的持久化队列
                .durable("simple-queue")
                .build();
    }

    @Bean
    public 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 {

    @Bean
    public DirectExchange workQueueExchange(){
        return ExchangeBuilder
                .directExchange("work-queue-exchange")
                .durable(true)
                .build();
    }

    @Bean
    public Queue workQueue(){
        return QueueBuilder
                .durable("work-queue")
                .build();
    }

    @Bean
    public 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 {

    @Bean
    public Queue publishSubscribeQueue1() {
        return new Queue("publish-subscribe-queue-1");
    }

    @Bean
    public Queue publishSubscribeQueue2() {
        return new Queue("publish-subscribe-queue-2");
    }

    @Bean
    public Queue publishSubscribeQueue3() {
        return new Queue("publish-subscribe-queue-3");
    }

    @Bean
    public FanoutExchange publishSubscribeFanoutExchange() {
        return new FanoutExchange("publish-subscribe-fanout-exchange");
    }

    @Bean
    public Binding publishSubscribeBinding1() {
        return BindingBuilder
                .bind(publishSubscribeQueue1())
                .to(publishSubscribeFanoutExchange());
    }

    @Bean
    public Binding publishSubscribeBinding2() {
        return BindingBuilder
                .bind(publishSubscribeQueue2())
                .to(publishSubscribeFanoutExchange());
    }

    @Bean
    public 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 {

    @Bean
    public DirectExchange routingExchange() {
        return ExchangeBuilder
                .directExchange("routing-exchange")
                .durable(true)
                .build();
    }

    @Bean
    public Queue routingQueue1() {
        return QueueBuilder
                .durable("routing-queue-1")
                .build();
    }

    @Bean
    public Queue routingQueue2() {
        return QueueBuilder
                .durable("routing-queue-2")
                .build();
    }

    @Bean
    public Queue routingQueue3() {
        return QueueBuilder
                .durable("routing-queue-3")
                .build();
    }

    @Bean
    public Binding routingQueue1ToRoutingExchangeBinding() {
        return BindingBuilder
                .bind(routingQueue1())
                .to(routingExchange())
                .with("routing-key-1");
    }

    @Bean
    public Binding routingQueue2ToRoutingExchangeBinding() {
        return BindingBuilder
                .bind(routingQueue2())
                .to(routingExchange())
                .with("routing-key-2");
    }

    @Bean
    public 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 {

    @Bean
    public Queue topicQueue1() {
        return new Queue("topic-queue-1");
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue("topic-queue-2");
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topic-exchange");
    }

    @Bean
    public Binding bindingExchange() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("log.info.*");
    }

    @Bean
    public 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 {

    @Bean
    public Queue rpcQueue() {
        return new Queue("rpc-queue", true);
    }

    @Bean
    public DirectExchange rpcExchange() {
        return new DirectExchange("rpc-exchange");
    }

    @Bean
    public Binding rpcQueueToRpcExchangeBinding() {
        return BindingBuilder.bind(rpcQueue()).to(rpcExchange()).withQueueName();
    }

    @Bean
    public 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);
    }

}