在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);
}
}事务相关的解释：
-
setupConfirm方法： 该方法用于设置消息的 Confirm 机制，确保消息被正确地投递到 Exchange。如果开启了 Confirm 机制，当消息成功到达 Exchange 时，会触发 ConfirmCallback。在这里，它主要是设置PublisherCallbackChannel.RETURN_LISTENER_CORRELATION_KEY以及注册 ConfirmCallback。 -
isChannelLocallyTransacted方法： 判断当前的 Channel 是否是本地事务。如果是本地事务，会调用RabbitUtils.commitIfNecessary(channel)来提交事务。 -
sendToRabbit方法： 实际上将消息发送到 RabbitMQ 服务器。这一步之后，如果开启了 Confirm 机制，就会等待服务器的 Confirm。 -
RabbitUtils.commitIfNecessary(channel)方法： 这是 Spring AMQP 提供的工具方法，用于提交 Channel 上的本地事务。
为什么先发送消息再执行事务提交？
这是因为 RabbitMQ 的 Confirm 机制是异步的。当调用 sendToRabbit 发送消息后，消息会先被异步地发送到 RabbitMQ 服务器。如果开启了 Confirm 机制，你可以注册一个 ConfirmCallback，在消息成功到达 Exchange 时，RabbitMQ 会异步地调用你的 ConfirmCallback。这样就能确保消息被正确地发送到了 Exchange，然后再执行事务的提交。文章来源：https://www.toymoban.com/news/detail-804533.html
这样的设计是为了提高消息发送的吞吐量。如果等待每条消息都同步等待确认，会降低发送消息的速度。异步确认机制能够更好地适应高并发的场景，同时在 ConfirmCallback 中处理确认逻辑。文章来源地址https://www.toymoban.com/news/detail-804533.html
到了这里，关于rabbitmq | rabbitTemplate的convertAndSend部分源码解析的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
