spring中使用rabbitmq（spring-boot-starter-amqp）

记录在spring中如何使用amqp、一些典型的问题该怎么处理。顺带介绍一下rabbitmq中的概念和原理。

github.com/zhouruibest…

前提

创建一个Maven多模块项目，父配置文件pom.xml。其中 spring.boot.version 3.0.2, spring.cloud.version 2022.0.0, spring.cloud.alibaba.version 2022.0.0.0是官方指定的稳定版本组合。子模块可以继承这些配置，无需在各自的pom.xml中重复定义依赖版本。

<properties>
    <java.version>17</java.version>
    <maven.compiler.source>17</maven.compiler.source>
    <maven.compiler.target>17</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <spring.boot.version>3.0.2</spring.boot.version>
    <spring.cloud.version>2022.0.0</spring.cloud.version>
    <spring.cloud.alibaba.version>2022.0.0.0</spring.cloud.alibaba.version>
</properties>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>${spring.boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring.cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>${spring.cloud.alibaba.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

子模块的pom文件不用指定版本了：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <scope>compile</scope>
</dependency>

1. Direct、Topic、Fanout三种交换机的基本使用

使用配置类声明queue、exchange和binding。RabbitAdmin会识别到并在rabbitmq中创建出来。

@Configuration
public class RabbitMQConfig {
    private Logger logger = LoggerFactory.getLogger(RabbitMQConfig.class);

    // 配置Jackson消息转换器
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 配置RabbitTemplate并设置消息转换器
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置消息转换器
        rabbitTemplate.setMessageConverter(jsonMessageConverter());

        // 监控消息是否成功到达交换机
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                logger.info(\"消息已成功到达交换机: {}\", correlationData != null ? correlationData.getId() : \"未知ID\");
            } else {
                logger.error(\"消息到达交换机失败，原因: {}\", cause);
            }
        });

        // 用于监控消息路由失败的情况
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            logger.error(\"消息路由失败: 交换机={}, 路由键={}, 消息={}, 原因={}\", returnedMessage.getExchange(), returnedMessage.getRoutingKey(), returnedMessage.getMessage(), returnedMessage.getReplyText());
        });

        return rabbitTemplate;
    }

    // 配置AsyncRabbitTemplate并设置消息转换器 
    @Bean
    public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate rabbitTemplate) {
        // AsyncRabbitTemplate基于已配置好的RabbitTemplate构建
        // 会自动使用RabbitTemplate中设置的消息转换器
        AsyncRabbitTemplate asyncTemplate = new AsyncRabbitTemplate(rabbitTemplate);
        return asyncTemplate;
    }

    /*
    每一类Exchange的演示都通过单独的配置类来声明
    Tips：1. 配置类及其内部静态类中的@Bean注解，都可以被Spring Boot扫描到
          2. 我们只是声明了这些Queue、Exchange、Binding，RabbitAdmin 初始化的时候会从 spring 容器
             里取出所有的交换器 bean, 队列 bean, Binding Bean然后创建到RabbitMQ中
     */

    /**
     * Direct Exchange 示例的配置类
     */
    public static class DirectExchangeDemoConfiguration {

        @Bean
        public Queue queue0() {
            return new Queue(MessageForDirectExchange.QUEUE_NAME,
                    true, // durable: 是否持久化到磁盘，当 RabbitMQ 重启后，仍然存在
                    false, // exclusive: 是否排它，队列只对它的连接可见
                    false); // autoDelete: 当没有消费者时，自动删除
        }

        @Bean
        public DirectExchange exchange0() {
            return new DirectExchange(MessageForDirectExchange.EXCHANGE_NAME,
                    true,  // durable: 持久化到磁盘，当 RabbitMQ 服务重启后，该交换机会保留
                    false);  // autoDelete: 当最后一个绑定到该交换机的队列/交换机被解绑后，交换机会被自动删除
        }

        @Bean
        public Binding binding0() {
            return BindingBuilder.bind(queue0()).to(exchange0()).with(MessageForDirectExchange.ROUTING_KEY);
        }
    }

    /**
     * Topic Exchange 示例的配置类
     */
    public static class TopicExchangeDemoConfiguration {

        @Bean
        public Queue queue1() {
            return new Queue(MessageForTopicExchange.QUEUE_NAME,
                    true, // durable
                    false, // exclusive
                    false); // autoDelete
        }

        @Bean
        public TopicExchange exchange1() {
            return new TopicExchange(MessageForTopicExchange.EXCHANGE_NAME,
                    true,  // durable
                    false);  // autoDelete
        }

        @Bean
        public Binding binding1() {
            return BindingBuilder.bind(queue1()).to(exchange1()).with(MessageForTopicExchange.ROUTING_KEY);
        }

    }

    /**
     * Fanout Exchange 示例的配置类
     */
    public static class FanoutExchangeDemoConfiguration {

        @Bean
        public Queue queueA() {
            return new Queue(MessageForFanoutExchange.QUEUE_NAMEA,
                    true, // durable
                    false, // exclusive
                    false); // autoDelete
        }

        @Bean
        public Queue queueB() {
            return new Queue(MessageForFanoutExchange.QUEUE_NAMEB,
                    true, // durable
                    false, // exclusive
                    false); // autoDelete
        }

        @Bean
        public FanoutExchange exchange2() {
            return new FanoutExchange(MessageForFanoutExchange.EXCHANGE_NAME,
                    true,  // durable
                    false);  // autoDelete
        }

        @Bean
        public Binding bindingA() {
            return BindingBuilder.bind(queueA()).to(exchange2());
        }

        @Bean
        public Binding bindingB() {
            return BindingBuilder.bind(queueB()).to(exchange2());
        }
    }
}

RabbitTemplate 和 AsyncRabbitTemplate

RabbitTemplate 的应用场景：

• 同步发送消息
• 简单消息发送，对性能要求不高
• 不需要接收方接收消息后、将返回消息写到另一个queue以返回给调用方

RabbitTemplate 和 AsyncRabbitTemplate两者的确认机制相同, 都是通过 ConfirmCallback 和 ReturnsCallback 监控消息是否到达交换机和路由失败的情况。

AsyncRabbitTemplate 的设计初衷是支持异步请求-响应模式（RPC），重点提供了 convertSendAndReceive 方法。如果只是需要异步发送消息而不关心响应，可以直接使用 RabbitTemplate 的 convertAndSend 方法，并通过异步任务（如 Spring 的 @Async ）实现非阻塞.

通过配置类中的静态配置类声明

• 配置类及其内部静态类中的@Bean注解，都可以被Spring Boot扫描到
• 我们只是声明了这些Queue、Exchange、Binding，RabbitAdmin（Spring AMQP 提供的核心组件） 初始化的时候会从 spring 容器里取出所有的exchange bean, queue bean, binding bean然后创建到RabbitMQ中。

创建queue的参数

消息状态管理

• 未确认的消息：如果消息被消费但未确认，重启后可能会重新入队（取决于 requeue 设置）
• 已确认的消息：已确认的消息不会在RabbitMQ重启后恢复，因为 RabbitMQ 会删除这些消息
• 消息 TTL：如果消息设置了 TTL，需确保 TTL 时间足够长，避免在RabbitMQ重启前过期

可能导致消息不可见的情况

• 消息未持久化：即使队列持久化，如果消息未设置 deliveryMode=2 ，重启后消息内容会丢失
• 消息已确认：已确认的消息会被 RabbitMQ 删除，重启后不会恢复
• 消息已过期：如果消息在重启前已过期，且未配置死信队列，消息会被丢弃
• 队列非持久化：如果队列未设置 durable=true ，重启后队列和消息都会丢失

消息的可见性与状态的关系

DirectExchange

待发送的消息

@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageForDirectExchange {
    public static final String QUEUE_NAME = \"QUEUE0\";
    public static final String EXCHANGE_NAME = \"EXCHANGE0\";
    public static final String ROUTING_KEY = \"ROUTING_KEY0\";

    private String message;
}

生产者

@Component
@AllArgsConstructor
public class ProducerForDirectExchange {

    private final RabbitTemplate rabbitTemplate;

    public void syncSend(String msg) {
        var message = new MessageForDirectExchange();
        message.setMessage(msg);
        // 同步发送消息
        rabbitTemplate.convertAndSend(
                MessageForDirectExchange.EXCHANGE_NAME, // exchange
                MessageForDirectExchange.ROUTING_KEY, // routingKey
                message); // message
    }

    public void syncSendDefault(String msg) {
        MessageForDirectExchange message = new MessageForDirectExchange();
        message.setMessage(msg);
        // 同步发送消息, 使用默认交换机.默认的交换机连接了每一个队列并使用队列名作为路由键
        rabbitTemplate.convertAndSend(
                MessageForDirectExchange.QUEUE_NAME, // routingKey
                message); // message
    }

    @Async
    public ListenableFuture asyncSend(String msg) {
        try {
            // 发送消息
            this.syncSend(msg);
            this.syncSendDefault(msg + \", 使用默认交换机和路由键\");
            // 返回成功的 Future
            return AsyncResult.forValue(null);
        } catch (Throwable ex) {
            // 返回异常的 Future
            return AsyncResult.forExecutionException(ex);
        }
    }
}

• syncSend 最通用的使用场景，同步发送一条简单消息，发送到RabbitMQ就算成功
• asyncSend 异步发送消息场景，在一个单独的线程中发送、不阻塞当前的线程。对结果也不在意，AsyncResult.forValue(null)创建并返回一个已完成状态的Future对象。

消费者

消费者仅仅将消息打印一下：

public class ConsumerForDirectExchange {
    private Logger logger = LoggerFactory.getLogger(ConsumerForDirectExchange.class);

    //@RabbitHandler
    @RabbitListener(queues = MessageForDirectExchange.QUEUE_NAME)
    public void onMessage(Message message) {
        logger.info(\"接收到消息: \" + message);
        logger.info(\"消息头: {}\", message.getMessageProperties().getHeaders());
        // 打印消息体（原始字节流）
        logger.info(\"消息体: {}\", new String(message.getBody()));
        // 打印 content-type
        logger.info(\"content-type: {}\", message.getMessageProperties().getContentType());
    }
}

测试用例

@SpringBootTest(classes = Demo0Application.class)
public class DirectExchangeTest {
    private Logger logger = LoggerFactory.getLogger(DirectExchangeTest.class);

    @Autowired
    private ProducerForDirectExchange producer;

    @Test
    public void testSyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.syncSend(\"66666666\");
        logger.info(\"[testSyncSend][发送编号：[{}] 发送成功]\", id);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

    @Test
    public void testAsyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.asyncSend(\"这是一条direct exchange的消息\").addCallback(new ListenableFutureCallback() {

            @Override
            public void onFailure(Throwable e) {
                logger.info(\"[testASyncSend][发送编号：[{}] 发送异常]]\", id, e);
            }

            @Override
            public void onSuccess(Void aVoid) {
                logger.info(\"[testASyncSend][发送编号：[{}] 发送成功]\", id);
            }

        });
        logger.info(\"[testASyncSend][发送编号：[{}] 调用完成]\", id);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }
}

TopicExchange

待发送的消息

@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageForTopicExchange {
    public static final String QUEUE_NAME = \"QUEUE1\";
    public static final String EXCHANGE_NAME = \"EXCHANGE1\";
    public static final String ROUTING_KEY = \"order.#\";

    private String message;
}

生产者

@Component
@AllArgsConstructor
public class ProducerForTopicExchange {
    private final RabbitTemplate rabbitTemplate;
    private final AsyncRabbitTemplate asyncRabbitTemplate;

    public void syncSend(String msg) {
        var message = new MessageForTopicExchange();
        message.setMessage(msg);
        // 同步发送消息，阻塞式，返回条件为消息发送到交换机
        rabbitTemplate.convertAndSend(
                MessageForTopicExchange.EXCHANGE_NAME, // exchange
                MessageForTopicExchange.ROUTING_KEY, // routingKey
                message); // message
    }

    public RabbitFuture