论spring事务下的消息发送

旧城等待， 2023-06-13 09:30 2阅读 0赞

最近一直疲于重构canal consumer的客户端，近期也会陆续分享下一些开发同学常见的问题。其实很多时候发现问题并快速的解决问题并不难，难得时候是否有引起你的深度思考。谈起spring事务下发MQ消息，一般人都是直接撸面条式的硬代码，也有些高级的玩法会采用spring的ApplicationEvent来拓展实现，而spring中的@TransactionEventListener则是最高端的玩法，其背后的原理并不复杂，采用了事务和回调。

而采用ApplicationEvent的拓展并非独创，思想源于spring-session-data-redis（它session的创建与销毁方面确实非常巧妙）,一般实现为：

@Data
 public abstract class MQEvent<T extends MqMetaInfo> extends ApplicationEvent {
 private String topic;
 private String tags;
 private String keys;
 private T body;
 private boolean needBodyJSONSerialize;
 private boolean areYouOK = true;
 }
 public abstract class MQApplicationListener<T extends MQEvent> implements ApplicationListener<T> {
 @Override
 public void onApplicationEvent(T event) {
 if (event.isAreYouOK()) {
 if (Strings.isNullOrEmpty(event.getTopic())) {
 event.setTopic(getTopic());
 }
 if (StringUtils.isBlank(event.getTags())) {
 event.setTags(getTopic());
 }
 getRocketMQProducer().sendMessage(event.getTopic(), event.getTags(), event.getKeys(), event.isNeedBodyJSONSerialize() ? JSON.toJSONString(event.getBody()) : String.valueOf(event.getBody()));
 }
 }
 protected abstract String getTopic();
 protected abstract String getTag();
 protected abstract RocketMQProducerUtil getRocketMQProducer();
 }
 @Component
 public class MQEventBus {
 @Autowired
 private ApplicationContext applicationContext;
 public void publishEvent(MQEvent event) {
 applicationContext.publishEvent(event);
 }
 public void publishEvents(MQEvent... events) {
 Assert.notNull(events, "publish Events is required.");
 Stream.of(events).forEach(x -> publishEvent(x));
 }
 }
 //通过API的MQEvent，MQApplicationListener来拓展
 @Component
 public class OrderMessageListener extends MQApplicationListener<OrderMessageListener.OrderCreateEvent> {
 @Value("${rocketmq.producer.orderCreateTopic:#{null}}")
 private String topic;
 @Value("${rocketmq.producer.orderCreateTag:#{null}}")
 private String tag;
 @Autowired
 private RocketMQProducerUtil rocketMQProducerUtil;
 @Override
 public RocketMQProducerUtil getRocketMQProducer() {
 return rocketMQProducerUtil;
 }
 @Override
 public String getTopic() {
 return topic;
 }
 @Override
 protected String getTag() {
 return tag;
 }
 public static class OrderCreateEvent extends MQEvent<ReqEccBmsOrderMessageVO> {
 public OrderCreateEvent(ReqEccBmsOrderMessageVO body) {
 this(null, null, body, true);
 }
 public OrderCreateEvent(String tags, String keys, ReqEccBmsOrderMessageVO body, boolean needBodyJSONSerialize) {
 super("OrderCreateEvent", null, tags, keys, body, needBodyJSONSerialize);
 }
 }
 }
 //示例代码1
 @Transactional(rollbackFor = Exception.class)
 public void saveOrder(List<BaseOrder> orders, List<BaseOrderDetail> orderDetails) {
 orderService.saveBatch(orders);
 orderDetailService.saveBatch(orderDetails);
 mqEventBus.publishEvent(...)
 }

实现还比较简单吧，但还需要考虑2个问题：

1.  是否需要大事务，一般情况下数据库操作和发MQ消息都会在spring事务中完成，见示例代码1。虽说这样的用法随处可见，但有个问题需要思考：mqEventBus.publishEvent()为同步阻塞时，换言之在这期间当前的事务不会提交的（这边不就是长事务么，说好的快进快出呢），假如publishEvent延迟严重，那mysql性能消耗就比较堪忧了；
2.  同步还是异步，异步最大的好处在于对消息的生产者与消费进行解耦，属于关注点分离范畴，它的好处就在于可补偿，最终一致性，性能更好。

接下来我们看下spring这块的实现，在**Spring4.2以前**：

刚刚讲过**事务不会提交的问题**也是可以解决的！可以在commit/rollback之后触发回调，关注TransactionSynchronization接口的afterCompletion方法即可，但使用它需要特别注意它会影响连接的释放（换言之，只有执行玩才会将会连结归还连接池，如果有大量耗时操作，那数据库的连接池就game over了）。

public interface TransactionSynchronization extends Flushable {
 
 	/** Completion status in case of proper commit. */
 	int STATUS_COMMITTED = 0;
 	/** Completion status in case of proper rollback. */
 	int STATUS_ROLLED_BACK = 1;
 	/** Completion status in case of heuristic mixed completion or system errors. */
 	int STATUS_UNKNOWN = 2;
 	/**
 	 * Suspend this synchronization.
 	 * Supposed to unbind resources from TransactionSynchronizationManager if managing any.
 	 * @see TransactionSynchronizationManager#unbindResource
 	 */
 	default void suspend() {
 	}
 	/**
 	 * Resume this synchronization.
 	 * Supposed to rebind resources to TransactionSynchronizationManager if managing any.
 	 * @see TransactionSynchronizationManager#bindResource
 	 */
 	default void resume() {
 	}
 
 	/**
 	 * Flush the underlying session to the datastore, if applicable:
 	 * for example, a Hibernate/JPA session.
 	 * @see org.springframework.transaction.TransactionStatus#flush()
 	 */
 	@Override
 	default void flush() {
 	}
 	/**
 	 * Invoked before transaction commit (before "beforeCompletion").
 	 * Can e.g. flush transactional O/R Mapping sessions to the database.
 	 */
 	default void beforeCommit(boolean readOnly) {
 	}
 
 	/**
 	 * Invoked before transaction commit/rollback.
 	 * Can perform resource cleanup before transaction completion.
 	 */
 	default void beforeCompletion() {
 	}
 
 	/**
 	 * Invoked after transaction commit. Can perform further operations right
 	 * after the main transaction has successfully committed.
 	 */
 	default void afterCommit() {
 	}
 	/**
 	 * Invoked after transaction commit/rollback.
 	 * Can perform resource cleanup after transaction completion.
 	 * NOTE: The transaction will have been committed or rolled back already,
 	 * but the transactional resources might still be active and accessible. 
 	 */
 	default void afterCompletion(int status) {
 	}
 }
 //示例代码
 @Transactional(rollbackFor = Exception.class)
 public void saveOrder (List < BaseOrder > orders, List < BaseOrderDetail > orderDetails){
 orderService.saveBatch(orders);
 orderDetailService.saveBatch(orderDetails);
 if (TransactionSynchronizationManager.isActualTransactionActive()) {
 TransactionSynchronizationManager.registerSynchronization(
 new TransactionSynchronizationAdapter() {
 @Override
 public void afterCommit() {
 mqEventBus.publishEvent(...)
 }
 });
 } else {
 mqEventBus.publishEvent(...)
 }
 }
 //源码分析
 public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
 	@Override
 	public final void commit(TransactionStatus status) throws TransactionException {
 		if (status.isCompleted()) {
 			throw new IllegalTransactionStateException(
 					"Transaction is already completed - do not call commit or rollback more than once per transaction");
 		}
 
 		DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
 		if (defStatus.isLocalRollbackOnly()) {
 			if (defStatus.isDebug()) {
 				logger.debug("Transactional code has requested rollback");
 			}
 			processRollback(defStatus, false);
 			return;
 		}
 
 		if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
 			if (defStatus.isDebug()) {
 				logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
 			}
 			processRollback(defStatus, true);
 			return;
 		}
 		//真正开始提交事务
 		processCommit(defStatus);
 	}
 	/**
 	 * Process an actual commit.
 	 * Rollback-only flags have already been checked and applied.
 	 * @param status object representing the transaction
 	 * @throws TransactionException in case of commit failure
 	 */
 	private void processCommit(DefaultTransactionStatus status) throws TransactionException {
 		try {
 			boolean beforeCompletionInvoked = false;
 
 			try {
 				boolean unexpectedRollback = false;
 				prepareForCommit(status);
 				triggerBeforeCommit(status);
 				triggerBeforeCompletion(status);
 				beforeCompletionInvoked = true;
 
 				if (status.hasSavepoint()) {
 					if (status.isDebug()) {
 						logger.debug("Releasing transaction savepoint");
 					}
 					unexpectedRollback = status.isGlobalRollbackOnly();
 					status.releaseHeldSavepoint();
 				}
 				else if (status.isNewTransaction()) {
 					if (status.isDebug()) {
 						logger.debug("Initiating transaction commit");
 					}
 					unexpectedRollback = status.isGlobalRollbackOnly();
 					doCommit(status);//执行事务提交
 				}
 				else if (isFailEarlyOnGlobalRollbackOnly()) {
 					unexpectedRollback = status.isGlobalRollbackOnly();
 				}
 
 				// Throw UnexpectedRollbackException if we have a global rollback-only
 				// marker but still didn't get a corresponding exception from commit.
 				if (unexpectedRollback) {
 					throw new UnexpectedRollbackException(
 							"Transaction silently rolled back because it has been marked as rollback-only");
 				}
 			}
 			catch (UnexpectedRollbackException ex) {
 				// can only be caused by doCommit
 				triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
 				throw ex;
 			}
 			catch (TransactionException ex) {
 				// can only be caused by doCommit
 				if (isRollbackOnCommitFailure()) {
 					doRollbackOnCommitException(status, ex);
 				}
 				else {
 					triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
 				}
 				throw ex;
 			}
 			catch (RuntimeException | Error ex) {
 				if (!beforeCompletionInvoked) {
 					triggerBeforeCompletion(status);
 				}
 				doRollbackOnCommitException(status, ex);
 				throw ex;
 			}）
 			// Trigger afterCommit callbacks, with an exception thrown there
 			// propagated to callers but the transaction still considered as committed.
 			try {
 				triggerAfterCommit(status);
 			}
 			finally {
 				//执行的回调（发消息）
 				triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
 			}
 		}
 		finally {
 			//此次归还connection
 			cleanupAfterCompletion(status);
 		}
 	}
 	private void cleanupAfterCompletion(DefaultTransactionStatus status) {
 		status.setCompleted();
 		if (status.isNewSynchronization()) {
 			TransactionSynchronizationManager.clear();
 		}
 		if (status.isNewTransaction()) {//此次非常重要，会调用子类DataSourceTransactionManager的doCleanupAfterCompletion
 			doCleanupAfterCompletion(status.getTransaction());
 		}
 		if (status.getSuspendedResources() != null) {
 			if (status.isDebug()) {
 				logger.debug("Resuming suspended transaction after completion of inner transaction");
 			}
 			Object transaction = (status.hasTransaction() ? status.getTransaction() : null);
 			resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());
 		}
 	}
 }
 public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
 		implements ResourceTransactionManager, InitializingBean {
 	@Override
 	protected void doCleanupAfterCompletion(Object transaction) {
 		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
 
 		// Remove the connection holder from the thread, if exposed.
 		if (txObject.isNewConnectionHolder()) {//释放连接
 			TransactionSynchronizationManager.unbindResource(obtainDataSource());
 		}
 
 		// Reset connection.
 		Connection con = txObject.getConnectionHolder().getConnection();
 		try {
 			if (txObject.isMustRestoreAutoCommit()) {
 				con.setAutoCommit(true);
 			}
 			DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
 		}
 		catch (Throwable ex) {
 			logger.debug("Could not reset JDBC Connection after transaction", ex);
 		}
 
 		if (txObject.isNewConnectionHolder()) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
 			}
 			//释放连接
 			DataSourceUtils.releaseConnection(con, this.dataSource);
 		}
 		//清理资源
 		txObject.getConnectionHolder().clear();
 	}
 }

从Spring4.2开始，引入了@TransactionEventListener，它的原理与上面是类似的，只不过用法上更加简单更友好。首先借助了@TransactionalEventListener 注解（你可以认为她是@EventListener 的派生类），借助了EventListenerMethodProcessor 的针对每个方法生成一个ApplicationListenerMethodTransactionalAdapter 对象，而回掉实现的也比较巧妙！

回调的对象是TransactionSynchronizationEventAdapter（是TransactionSynchronization的一个实现类），每次调用onApplicationEvent时先实例化一个TransactionSynchronizationEventAdapter并通过TransactionSynchronizationManager.registerSynchronization添加，而ApplicationListenerMethodTransactionalAdapter既是ApplicationEvent的Listener，又是TransactionSynchronization真正被调用的类（业务方法）。

/* @since 4.2
 * @see EventListenerMethodProcessor
 */
 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 public @interface EventListener {
 	@AliasFor("classes")
 	Class<?>[] value() default {};
 	@AliasFor("value")
 	Class<?>[] classes() default {};
 	/**
 	 * Spring Expression Language (SpEL) attribute used for making the
 	 * event handling conditional.
 	 * Default is {@code ""}, meaning the event is always handled.
 	 * The SpEL expression evaluates against a dedicated context that
 	 * provides the following meta-data:
 	 * <ul>
 	 * <li>{@code #root.event}, {@code #root.args} for
 	 * references to the {@link ApplicationEvent} and method arguments
 	 * respectively.</li>
 	 * <li>Method arguments can be accessed by index. For instance the
 	 * first argument can be accessed via {@code #root.args[0]}, {@code #p0}
 	 * or {@code #a0}. Arguments can also be accessed by name if that
 	 * information is available.</li>
 	 * </ul>
 	 */
 	String condition() default "";
 }
 /*
 * @since 4.2
 */
 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @EventListener
 public @interface TransactionalEventListener {
 	/**
 	 * Phase to bind the handling of an event to.
 	 * The default phase is {@link TransactionPhase#AFTER_COMMIT}.
 	 * If no transaction is in progress, the event is not processed at
 	 * all unless {@link #fallbackExecution} has been enabled explicitly.
 	 */
 	TransactionPhase phase() default TransactionPhase.AFTER_COMMIT;
 	/**
 	 * Whether the event should be processed if no transaction is running.
 	 */
 	boolean fallbackExecution() default false;
 	@AliasFor(annotation = EventListener.class, attribute = "classes")
 	Class<?>[] value() default {};
 	@AliasFor(annotation = EventListener.class, attribute = "classes")
 	Class<?>[] classes() default {};
 	String condition() default "";
 }
 //示例代码
 @Component
 public class OrderMessageListener {
 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
 public void hanldeOrderCreatedEvent(OrderCreateEvent event) {
 ...
 }
 }
 @Transactional(rollbackFor = Exception.class)
 public void saveOrder(List<BaseOrder> orders, List<BaseOrderDetail> orderDetails) {
 orderService.saveBatch(orders);
 orderDetailService.saveBatch(orderDetails);
 mqEventBus.publishEvent(...)
 }
 /* @since 4.2
 */
 public interface EventListenerFactory {
 }
 //SmartInitializingSingleton是bean生命周期里很重要的拓展，所有单例bean都初始化完成以后， 容器会回调该接口的方法
 public class EventListenerMethodProcessor implements SmartInitializingSingleton, ApplicationContextAware {
 	@Override
 	public void afterSingletonsInstantiated() {
 ...
 	}
 }
 //很明显，这个是Listener的工厂类
 public class TransactionalEventListenerFactory implements EventListenerFactory, Ordered {
 	@Override
 	public boolean supportsMethod(Method method) {
 		return AnnotatedElementUtils.hasAnnotation(method, TransactionalEventListener.class);
 	}
 
 	@Override
 	public ApplicationListener<?> createApplicationListener(String beanName, Class<?> type, Method method) {
 		return new ApplicationListenerMethodTransactionalAdapter(beanName, type, method);
 	}
 
 }
 //源码解析
 /**
 * 它就是一个ApplicationListener的实现类，其实Event啥都不判断，最终还是调用的applicationContext.publishEvent
 */
 class ApplicationListenerMethodTransactionalAdapter extends ApplicationListenerMethodAdapter {
 
 	private final TransactionalEventListener annotation;
 
 
 	public ApplicationListenerMethodTransactionalAdapter(String beanName, Class<?> targetClass, Method method) {
 		super(beanName, targetClass, method);
 		TransactionalEventListener ann = AnnotatedElementUtils.findMergedAnnotation(method, TransactionalEventListener.class);
 		if (ann == null) {
 			throw new IllegalStateException("No TransactionalEventListener annotation found on method: " + method);
 		}
 		this.annotation = ann;
 	}
 
 
 	@Override
 	public void onApplicationEvent(ApplicationEvent event) {
 //这里是精髓，向当前的TransactionSynchronizationManager中注入synchronizations（TransactionSynchronization）
 		if (TransactionSynchronizationManager.isSynchronizationActive()) {
 			TransactionSynchronization transactionSynchronization = createTransactionSynchronization(event);
 			TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);
 		}
 		else if (this.annotation.fallbackExecution()) {
 			if (this.annotation.phase() == TransactionPhase.AFTER_ROLLBACK && logger.isWarnEnabled()) {
 				logger.warn("Processing " + event + " as a fallback execution on AFTER_ROLLBACK phase");
 			}
 			processEvent(event);
 		}
 		else {
 			// No transactional event execution at all
 			if (logger.isDebugEnabled()) {
 				logger.debug("No transaction is active - skipping " + event);
 			}
 		}
 	}
 
 	private TransactionSynchronization createTransactionSynchronization(ApplicationEvent event) {
 		return new TransactionSynchronizationEventAdapter(this, event, this.annotation.phase());
 	}
 
 	//一个TransactionSynchronization的实例，等待被回调
 	private static class TransactionSynchronizationEventAdapter extends TransactionSynchronizationAdapter {
 
 		private final ApplicationListenerMethodAdapter listener;
 
 		private final ApplicationEvent event;
 
 		private final TransactionPhase phase;
 
 		public TransactionSynchronizationEventAdapter(ApplicationListenerMethodAdapter listener,
 				ApplicationEvent event, TransactionPhase phase) {
 
 			this.listener = listener;
 			this.event = event;
 			this.phase = phase;
 		}
 
 		@Override
 		public int getOrder() {
 			return this.listener.getOrder();
 		}
 
 		@Override
 		public void beforeCommit(boolean readOnly) {
 			if (this.phase == TransactionPhase.BEFORE_COMMIT) {
 				processEvent();
 			}
 		}
 
 		@Override
 		public void afterCompletion(int status) {
 			if (this.phase == TransactionPhase.AFTER_COMMIT && status == STATUS_COMMITTED) {
 				processEvent();
 			}
 			else if (this.phase == TransactionPhase.AFTER_ROLLBACK && status == STATUS_ROLLED_BACK) {
 				processEvent();
 			}
 			else if (this.phase == TransactionPhase.AFTER_COMPLETION) {
 				processEvent();
 			}
 		}
 
 		protected void processEvent() {
 			this.listener.processEvent(this.event);
 		}
 	}
 
 }
 /**
 * 处理xml上的实现，在<tx:annotation-driven/>时触发，基本上支持spring 2.0，就注入了这个bean TransactionalEventListenerFactory
 */
 class AnnotationDrivenBeanDefinitionParser implements BeanDefinitionParser {
 	/**
 	 * Parses the {@code <tx:annotation-driven/>} tag. Will
 	 * {@link AopNamespaceUtils#registerAutoProxyCreatorIfNecessary register an AutoProxyCreator}
 	 * with the container as necessary.
 	 */
 	@Override
 	@Nullable
 	public BeanDefinition parse(Element element, ParserContext parserContext) {
 		registerTransactionalEventListenerFactory(parserContext);
 		String mode = element.getAttribute("mode");
 		if ("aspectj".equals(mode)) {
 			// mode="aspectj"
 			registerTransactionAspect(element, parserContext);
 			if (ClassUtils.isPresent("javax.transaction.Transactional", getClass().getClassLoader())) {
 				registerJtaTransactionAspect(element, parserContext);
 			}
 		}
 		else {
 			// mode="proxy"
 			AopAutoProxyConfigurer.configureAutoProxyCreator(element, parserContext);
 		}
 		return null;
 	}
 	//开始注入TransactionalEventListenerFactory Listener
 	private void registerTransactionalEventListenerFactory(ParserContext parserContext) {
 		RootBeanDefinition def = new RootBeanDefinition();
 		def.setBeanClass(TransactionalEventListenerFactory.class);
 		parserContext.registerBeanComponent(new BeanComponentDefinition(def,
 				TransactionManagementConfigUtils.TRANSACTIONAL_EVENT_LISTENER_FACTORY_BEAN_NAME));
 	}
 }
 /**
 * spring boot中也同样比较简单
 */
 @Configuration
 public abstract class AbstractTransactionManagementConfiguration implements ImportAware {
 	@Bean(name = TransactionManagementConfigUtils.TRANSACTIONAL_EVENT_LISTENER_FACTORY_BEAN_NAME)
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 	public static TransactionalEventListenerFactory transactionalEventListenerFactory() {
 		return new TransactionalEventListenerFactory();
 	}
 }

spring在设计工业上有太多太多需要我们学习的地方，比如注解的派生性、接口的生命周期及回调，很多时候你会发现其实context包变动很小，却可以支持很多方面的拓展，开闭原则用到了极致。![95242448][]

[95242448]: https://blog.csdn.net/alex_xfboy/article/details/95242448