@Bean
public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) {
    // channel开启事务支持
    rabbitTemplate.setChannelTransacted(true);
    return new RabbitTransactionManager(connectionFactory);
}

//使用spring事务，必须使用该注解
@Transactional
public void send(String messageStr, String routeKey) {
    String messageId = String.valueOf(UUID.randomUUID());
    Body body = new Body();
    body.setCreateTime(DateTime.now());
    body.setMessage(messageStr);
    body.setMessageId(messageId);
    //将消息携带绑定键值，发送到交换机
    rabbitTemplate.convertAndSend("testTopic", routeKey, body);
    System.out.println(rabbitTemplate.isChannelTransacted());
}

protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
    TransactionAttributeSource tas = this.getTransactionAttributeSource();
    TransactionAttribute txAttr = tas != null ? tas.getTransactionAttribute(method, targetClass) : null;
    PlatformTransactionManager tm = this.determineTransactionManager(txAttr);
    String joinpointIdentification = this.methodIdentification(method, targetClass, txAttr);
    Object result;
    if (txAttr != null && tm instanceof CallbackPreferringPlatformTransactionManager) {
        //该部分代码比较复杂，没有贴出，可以自行查看，其实就是我没看懂，所以没贴
    } else {
        TransactionAspectSupport.TransactionInfo txInfo = this.createTransactionIfNecessary(tm, txAttr, joinpointIdentification);

        try {
            //这里代表我们实际的方法，即加上@Transactional注解的方法，spring在注册bean时，会扫描包含@Transactional注解的方法，然后生成代理方法，即aop拦截
            result = invocation.proceedWithInvocation();
        } catch (Throwable var17) {
            //如果发生了报错，则回滚，调用PlatformTransactionManager的rollback()，最终调用channel的txRollback()
            this.completeTransactionAfterThrowing(txInfo, var17);
            throw var17;
        } finally {
            this.cleanupTransactionInfo(txInfo);
        }
        //如果执行成功，则提交，调用PlatformTransactionManager的commit()，最终调用channel的txCommit()
        this.commitTransactionAfterReturning(txInfo);
        return result;
    }
}

protected void completeTransactionAfterThrowing(@Nullable TransactionAspectSupport.TransactionInfo txInfo, Throwable ex) {
if (txInfo != null && txInfo.getTransactionStatus() != null) {
    if (this.logger.isTraceEnabled()) {
        this.logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "] after exception: " + ex);
    }
    //这里主要是判断事务传播机制，以及错误类型是否需要回滚，错误类型由@Transactional注解的rollbackFor属性决定，默认RuntimeException
    if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
        try {
            txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
        } catch (TransactionSystemException var6) {
            this.logger.error("Application exception overridden by rollback exception", ex);
            var6.initApplicationException(ex);
            throw var6;
        } catch (Error | RuntimeException var7) {
            this.logger.error("Application exception overridden by rollback exception", ex);
            throw var7;
        }
    } else {
        try {
            txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
        } catch (TransactionSystemException var4) {
            this.logger.error("Application exception overridden by commit exception", ex);
            var4.initApplicationException(ex);
            throw var4;
        } catch (Error | RuntimeException var5) {
            this.logger.error("Application exception overridden by commit exception", ex);
            throw var5;
        }
    }
}

}