【分布式事务】可靠消息最终一致性方案

旧城等待， 2022-06-10 08:21 257阅读 0赞

### 【回顾】 ###

####     在上篇博客中，我们了解了标准分布式事务解决方案，即全局事务。博客最后也总结到：由于其性能上的缺陷，对于高并发系统并不适用，所以一般采用柔性事务解决方案，本篇博客要介绍的就是柔性事务方案之一：可靠消息最终一致性方案。 ####

### 【产生背景】 ###

####     使用两阶段或三阶段提交协议完成分布式事务，一般来说性能较差，因为事务管理器需要在多个数据源之间进行多次等待。 ####

### 【简单实例】 ###

####     就拿互联网应用的用户注册功能来说，通常都有两个操作： ####

####         - 注册成功，保存用户信息； ####

####         - 注册成功，给用户对应的积分或成长值。 ####

####     如果是一个单体架构实现这个功能就非常简单，在一个本地事务里，往用户信息表中插入一条记录，并且往积分表中插入一条记录，提交事务就完成了。 ####

####     但如果是微服务架构，用户和积分通常是两个独立的服务，它们有各自的应用和数据库，那么本地事务就失效了。 ####

### 【解决方案】 ###

####     对于上述实例的问题，我们便可以采用事件机制和消息队列实现分布式事务，以确保消息的最终一致性。 ####

### 【应用实战】 ###

####     下面将结合代码，以用户签到获取对应积分的需求为例，看看如何具体实现。 ####

####     1. 在签到服务中，首先我们需要配置事件如下： ####

####     签到累积积分逻辑代码如下： ####

/** * 签到积累积分 * * @param customerId * @throws BusinessException */
        private void fireEventLoyalty(Long customerId) {
            try {
                LoyaltyBuildModel loyaltyBuildModel = new LoyaltyBuildModel();
                loyaltyBuildModel.setCustomerId(customerId);
                loyaltyBuildModel.setChannel("");
                loyaltyBuildModel.setEventId(LOYALTY_EVENT_SIGN);
                Map<String, Object> context = Maps.newHashMap();
                loyaltyBuildModel.setContext(context);
                String data = JsonMapper.toJson(loyaltyBuildModel);
                LOG.debug("fireEventLoyalty with data:{}", data);
                businessEventSource.fireEvent(new LoyaltyBuildEvent<String>(com.gac.loyalty.constant.EventConstant.LOYALTY_GET, this.getClass().getName(), data));
            } catch (Exception e) {
                LOG.error("fireEventLoyalty error:{}", e.getMessage());
            }
        }

####     2. 在积分服务中，同样先配置事件如下： ####

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd" default-lazy-init="true">
    
        <description>Event配置文件</description>
        <bean id="targetEventConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="${GAC_BROKER_URL}"/>
        </bean>
        <bean id="eventConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
            <property name="targetConnectionFactory" ref="targetEventConnectionFactory"/>
        </bean>
    
        
        <bean id="eventQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg value="${GAC_MQ_QUEUE_NAME}"/>
        </bean>
    
        
        
        <bean id="loyaltyBuildEventListener" class="com.gac.loyalty.service.business.listener.LoyaltyBuildEventListener">
            <property name="focus" value="${GAC_MQ_MONITOR_QUEUE_NAME}"/>
        </bean>
    
        <bean id="eventNotifier" class="com.gac.common.event.local.BusinessEventLocalNotifierImpl">
            <constructor-arg>
                <list>
                    <ref bean="loyaltyBuildEventListener"/>
                </list>
            </constructor-arg>
        </bean>
        
        <bean id="eventSource" class="com.gac.common.event.BusinessEventSource">
            <property name="notifier" ref="eventNotifier"/>
        </bean>
        
        <bean id="modelIndexMessageListener" class="com.gac.common.event.queue.BusinessModelEventMessageListener">
            <property name="eventSource" ref="eventSource"/>
        </bean>
        
        <bean id="jmsSeriesContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="eventConnectionFactory"/>
            <property name="destination" ref="eventQueueDestination"/>
            <property name="messageListener" ref="modelIndexMessageListener"/>
        </bean>
        
    
        
        <bean id="jmsEventTemplate" class="org.springframework.jms.core.JmsTemplate">
            <property name="connectionFactory" ref="eventConnectionFactory"/>
        </bean>
        
        <bean id="eventMessageProducer" class="com.gac.common.activemq.NotifyMessageProducer">
            <property name="jmsTemplate" ref="jmsEventTemplate"/>
            <property name="notifyQueue" ref="eventQueueDestination"/>
        </bean>
        
        <bean id="eventNotifierSend" class="com.gac.common.event.queue.BusinessEventQueueNotifier">
            <property name="messageProducer" ref="eventMessageProducer"/>
            <property name="queue" value="true"/>
        </bean>
        
        <bean id="businessEventSource" class="com.gac.common.event.BusinessEventSource">
            <property name="notifier" ref="eventNotifierSend"/>
        </bean>
        
    </beans>

####     积分服务事件监听器代码如下： ####

/** * 监听逻辑处理 * * @param businessEvent * @throws BusinessEventException */
        @Override
        public void handle(BusinessEvent<String> businessEvent) throws BusinessEventException {
            String data = businessEvent.getData();
            LOG.debug("LoyaltyBuildEventListener handle with:{}", data);
    
            if (StringUtils.isEmpty(data)) {
                LOG.error("LoyaltyBuildEventListener handled nothing, no data received.");
                return;
            }
            LoyaltyBuildModel loyaltyBuildModel = JSONObject.parseObject(data, new TypeReference<LoyaltyBuildModel>() {
            });
            if (loyaltyBuildModel == null) {
                LOG.error("LoyaltyBuildEventListener handled nothing, no data received.");
                return;
            }
    
            try {
                loyaltyBusinessService.updateLoyalty(loyaltyBuildModel.getCustomerId(), loyaltyBuildModel.getChannel(),
                        loyaltyBuildModel.getEventId(), loyaltyBuildModel.getContext());
            } catch (BusinessException e) {
                LOG.error("LoyaltyBuildEventListener error:{}", e.getMessage());
            }
        }

### 【实现原理】 ###

####     以上代码是实现的主要代码，其中有些公共代码的封装，可参考博客：[【java】事件与观察者模式][java]。 ####

####     通过以上配置，可以看到采用了JMS的点对点的消息模型，其特性包含如下几点： ####

####         - 保证了每个消息只有一个消费者； ####

####         - 发送者和接收者之间在时间上没有依赖性； ####

####         - 接收者在成功接收消息之后需向队列应答成功。 ####

####     在此方案中，采用点对点模型最重要的是它可以保证每个消息都可以被成功处理。加上消息中间件ActiveMQ的性能提升，便可以很好地处理微服务架构中分布式事务。 ####

### 【比较】 ###

####     和2PC同步事务相比，这种异步方式优势在于： ####

####         - 事务吞吐量大，因为不需要等待其他数据源响应。 ####

####         - 容错性好。签到服务在发送事件的时候，积分服务甚至可以不在运行。 ####

### 【总结】 ###