rabbitmq | rabbitTemplate的convertAndSend部分源码解析

在RabbitMQ中，事务是一种确保消息发送的可靠性的机制。Spring AMQP提供了对RabbitMQ事务的抽象，而

RabbitTemplate

作为Spring AMQP的核心组件，提供了许多简化消息发送的方法。在这篇博客中，我们将深入探讨RabbitMQ事务机制的源码实现，以及Spring封装的

RabbitTemplate

的使用。

RabbitMQ事务机制

RabbitMQ的事务机制基于AMQP协议提供的事务操作，主要涉及到三个关键方法(可看Channel提供的源码)：

txSelect

（启动事务）、

basicPublish

（消息发送）、

txCommit

（事务提交）和

txRollback

（事务回滚）。下面我们将分别解释它们在队列事务中的作用。

txSelect - 启动事务

public interface Channel extends ShutdownNotifier, AutoCloseable {
  
 txSelect();

 ..........
   
}

txSelect

方法用于启动一个新的事务。在事务开始后，所有后续的消息发送操作都将在这个事务中进行。这意味着，直到事务被提交或回滚之前，这些消息都不会真正地到达消息队列。

basicPublish - 消息发送

channel.basicPublish(exchange, routingKey, properties, body);

basicPublish

方法用于发送消息到消息队列。在事务中，所有的消息发送都会被暂时缓存起来，直到事务被提交。这样可以确保只有在事务提交后，所有的消息才会真正地被发送到消息队列。

txCommit - 提交事务

channel.txCommit();

txCommit

方法用于提交当前事务。一旦调用了

txCommit

，之前通过

basicPublish

发送的所有消息将被一起提交到消息队列中。

Spring RabbitTemplate事务源码解析

在Spring中，

RabbitTemplate

封装了RabbitMQ的事务操作，使得事务的管理更加方便。下面我们将重点关注

convertAndSend

方法和

MessageProperties

的源码层面，解释事务如何开启、生效和回滚。

    @Override
    public void convertAndSend(String exchange, String routingKey, final Object object,
            @Nullable CorrelationData correlationData) throws AmqpException {

        send(exchange, routingKey, convertMessageIfNecessary(object), correlationData);
    }
-----------------------------------------------------------------------------------

    @Override
    public void send(final String exchange, final String routingKey,
            final Message message, @Nullable final CorrelationData correlationData)
            throws AmqpException {
        execute(channel -> {
            doSend(channel, exchange, routingKey, message,
                    (RabbitTemplate.this.returnsCallback != null
                            || (correlationData != null && StringUtils.hasText(correlationData.getId())))
                            && isMandatoryFor(message),
                    correlationData);
            return null;
        }, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));
    }

convertAndSend

方法是

RabbitTemplate

提供的一个高级发送消息的方法，它会自动进行消息的序列化和转换。

这个方法的主要作用是通过

execute

方法执行一个

ChannelCallback

，而

doSend

方法则是真正执行消息发送的逻辑。

doSend

方法：

public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message,
        boolean mandatory, @Nullable CorrelationData correlationData) throws IOException {
    // ...（部分代码省略）
    
    // 设置 confirm 机制
    setupConfirm(channel, messageToUse, correlationData);
    
    // 如果有 userIdExpression，则设置 userId
    if (this.userIdExpression != null && messageProperties.getUserId() == null) {
        String userId = this.userIdExpression.getValue(this.evaluationContext, messageToUse, String.class);
        if (userId != null) {
            messageProperties.setUserId(userId);
        }
    }
    
    // 打印 debug 日志
    if (logger.isDebugEnabled()) {
        logger.debug("Publishing message [" + messageToUse
                + "] on exchange [" + exch + "], routingKey = [" + rKey + "]");
    }
    
    // 发送消息到 RabbitMQ 服务器
    sendToRabbit(channel, exch, rKey, mandatory, messageToUse);
    
    // 如果 channel 是本地事务，则提交事务
    if (isChannelLocallyTransacted(channel)) {
        // Transacted channel created by this template -> commit.
        RabbitUtils.commitIfNecessary(channel);
    }
}

事务相关的解释：

1. setupConfirm 方法： 该方法用于设置消息的 Confirm 机制，确保消息被正确地投递到 Exchange。如果开启了 Confirm 机制，当消息成功到达 Exchange 时，会触发 ConfirmCallback。在这里，它主要是设置 PublisherCallbackChannel.RETURN_LISTENER_CORRELATION_KEY 以及注册 ConfirmCallback。
2. isChannelLocallyTransacted 方法： 判断当前的 Channel 是否是本地事务。如果是本地事务，会调用 RabbitUtils.commitIfNecessary(channel) 来提交事务。
3. sendToRabbit 方法： 实际上将消息发送到 RabbitMQ 服务器。这一步之后，如果开启了 Confirm 机制，就会等待服务器的 Confirm。
4. RabbitUtils.commitIfNecessary(channel) 方法： 这是 Spring AMQP 提供的工具方法，用于提交 Channel 上的本地事务。

为什么先发送消息再执行事务提交？

这是因为 RabbitMQ 的 Confirm 机制是异步的。当调用

sendToRabbit

发送消息后，消息会先被异步地发送到 RabbitMQ 服务器。如果开启了 Confirm 机制，你可以注册一个 ConfirmCallback，在消息成功到达 Exchange 时，RabbitMQ 会异步地调用你的

ConfirmCallback

。这样就能确保消息被正确地发送到了 Exchange，然后再执行事务的提交。

这样的设计是为了提高消息发送的吞吐量。如果等待每条消息都同步等待确认，会降低发送消息的速度。异步确认机制能够更好地适应高并发的场景，同时在 ConfirmCallback 中处理确认逻辑。