0 引言

在互联网环境下，应用系统总会面对由于客户量爆炸式的增长而带来的系统压力，为了缓解系统压力目前普遍采用分布式系统架构，而进程间通信是分布式系统的核心技术, 目前广泛采用的进程间通信模式有:远程过程调用 (Remote Procedure Call, RPC)、远程方法调用 (Remote Method Invoke, RMI)、消息中间件 (Message-Oriented Middleware, MOM)^[1]以及流 (stream)。

消息传递机制可以避免通信阻塞, 增加系统的吞吐量, 以及解耦不同系统直接交互, 在不需要请求立即返回结果的场景下，这些特性带来了明显的通信优势。消息中间件的体系结构由消息生产者、消息中间件、消息消费者三个部分组成, 消息的生产者和消息的消费者只和消息中间件交互, 生产者和消费者不直接交互。

消息中间件的使用过程是, 当业务系统业务成功后, 创建消息, 然后向消息中间件发送消息, 消息中间件接收到消息并存储消息; 当消息中间件存在未被消费的消息时, 消息中间件向消费消息的业务系统推送消息。

从消息中间件使用过程看：一方面消息发送和消息消费通过网络交互, 网络不稳定会导致消息丢失或者重复；另一方面消息中间件消息推送机制不是实时的, 极端情况下会出现几天甚至几十天延迟, 这也会导致消息不一致。在使用消息中间件系统时，首先需要解决消息生产者和消息中间件之间以及消息中间件和消息消费者之间消息的一致性问题。这个一致性包括:1) 消息发送一致性, 即消息生产者业务成功发送消息到消息中间件, 消息中间件也能成功收到消息, 发送消息时网络出现问题或者消息中间件不可用, 则消息丢失; 2) 消息消费一致性, 即消息消费者只成功消费一次消息, 当消息消费者成功消费消息后, 向消息中间件发送确认消息时网络故障或者消息中间件不可用, 会导致消息消费者重复消费消息。

目前对消息中间件已经有很多研究文献, 文献[2]讨论了采用统一消息队列中间件软总线实现电力调度自动化系统, 设计和实现了多个应用子系统的集成, 但是并没有给出多个系统间消息一致性的解决方案, 这样不能满足分布式交易系统应用。

文献[3]指出传统二阶段提交不适合分布式复杂网络环境, “为了避免提交子事务等待时间过长, 而造成事务阻塞和对系统资源的浪费”, 把二阶段提交改成一阶段提交；但也指出一阶段提交是基于“全局事务最终能够正常提交的乐观想法上的”。文献指出传统二阶段分布式事务性能缺陷, 并通过采用消息机制解决一致性问题, 但存在两个问题:1) 不是所有场景都能改成消息机制来解决一致性；2) 改成消息机制后, 如何保障消息本身的一致性^[4]。

文献[5]中采用补偿的方法实现事务, 保证系统一致。其主要思想是当出现异常时, 会触发针对异常的一个回滚事件, 使系统恢复到执行前状态, 不会由于异常导致系统不一致；但文献中没有考虑网络异常情况, 没有加入幂等机制和不一致检查机制, 无法满足分布式交易系统一致性要求。

文献[6]较完整地分析了二阶段提交和三阶段提交保证分布式事务一致性带来的性能损耗, 采用非阻塞方式会提升性能；但没有考虑异常情况怎么处理, 无法满足事务的一致性。

文献[7-9]都指出了通过消息订阅/发布可实现松散耦合的、灵活的跨平台的通信机制的系统, 比传统的通信机制有更多的优点。

文献[10]指出消息中间件应用时, 遇到网络环境不稳定, 通过设计一个链路检测系统尽早发现网络问题, 但对于网络问题导致的系统不一致, 并没有给出方案。

文献[11]基于传输层UDP, 在消息中间件应用层设计了一种基于否定确认 (Negative ACKnowl-edgment, NACK) 策略, 在消息接收方通过检测报文丢失、乱序并提供消息重传机制, 保证传输的可靠性。不同功能单元采用双工热备实现节点可靠性。

文献[12]采用“Quorum-Based算法”与消息区域模型相结合的方法保证分布式消息中间件的数据一致性。其作用是保证多个消费节点状态一致, 其一致性类型属于最终一致性。

在文献[4, 13-14]中给出了分布式系统设计的总体原则, 根据CAP (Consistency, Availability, Partition tolerance) 理论^[15], 一个分布式系统不可能同时满足一致性、可用性和分区容错性这三个特性, 最多只能同时满足两个。尤其在文献[13]中, 提出了BASE (Basically Available, Soft state, Eventually consistent) 思想, 通过牺牲高一致性, 保证高可用性和分区容忍性, 这有很强的实践指导意义, 即分布式系统和传统系统有很大区别，以及系统设计时要懂得取舍。

在文献[16-17]中对分布式系统中数据一致性检测, 并且考虑了大数据的数据迁移以及检测的时效性, 但没有考虑分布式交易环境下一致性的检测和一致性保障。

针对上述相关研究, 总结如下:1) 在分布式场景中肯定了消息中间件可以解耦交互的系统, 提高系统的吞吐量的作用；2) 在保障分布式系统一致性上从多个方面进行了关注；3) 对分布式系统的设计给出了CAP原则；4) 关注了事务补偿机制在分布式事务中保持最终一致性的作用。

在现有的研究中还存在以下问题:1) 没有认真分析不同的分布式应用场景, 并且针对不同场景给出不同的一致性解决方案, 两阶段提交措施不适合高并发场景；2) 缺少最终一致性完整的解决方案, 分布式交易系统的一致性需要考虑多个方面, 譬如性能、可靠性, 从不一致的检测到实现一致性的多个环节；3) 没有指出产生不一致消息的时机，具体的时机可能是消息生产时、消息传递时、消息消费时, 造成不一致的原因可能是网络异常造成的, 可能是消息中间件延迟造成的, 或者可能是生产端接收端异常造成消息和业务的不一致, 对问题认识越深刻才能找到好的解决办法。

本文的主要工作:

1) 在分布式交易环境下, 提出如何采用消息通信方式来保障分布式系统间数据的最终一致性的完整方法, 包括：采用本地事务记录消息生产和消费, 采用后台进程比较数据发现异常, 采用幂等机制满足异常重试, 采用补偿措施进行异常回滚。通过这些方法形成分布式交易系统最终一致性完整解决方案。

2) 对消息通信方式和RPC通信方式的性能进行对比分析实验, 指出了在分布式高并发环境中消息通信方式性能上具有优势。

3) 采用软件工程化的方法进行关注点分离, 把一致性保障措施设计为一个平台, 有利于软件复用, 提高经济效益。

1 一致性保障层体系结构设计

分布式系统虽然有高性能的优点, 但同时也带来了问题, 如一致性降低等问题, 在交易系统中不允许存在不一致。为了揭示这个问题, 先从一个分布式应用场景入手分析导致不一致的原因, 然后给出解决不一致的方法机制, 最后设计出对应的软件体系结构。

1.1 消息通信的分布式系统应用场景分析

一个常见的场景是客户在网站注册, 注册成功则向这个用户送积分和现金抵用券, 最后再短信通知客户, 采用序列图^[18]表示注册过程，如图 1所示。注册、送积分、发短信和送抵用券分属不同系统, 系统之间通信采用同步方式或者异步方式。注册过程则不依赖其他三个过程。

在互联网应用环境中, 客户注册是个高并发应用场景，为了满足并发要求会把这个业务拆分成多个子业务分别实现在不同的系统中，形成分布式系统。在高并发环境下, 分布式系统一般采用异步的消息机制作为系统间的通信方式, 通过消息中间件解耦^[19]注册和其他三个过程, 解耦可以提高注册的并发度, 消息中间件适合异步通信场景, 改进后如图 2所示。注册作为消息生产者, 其他三个过程作为消息消费者, 如果不做其他工作, 在消息的生产和消费过程中不能保证一致, 会给用户带来不好的体验。

1.2 一致性保障机制设计

从图 2可以抽象出这样的分布式系统通信是由消息提供者、消息消费者、消息中间件和网络组成。根据消息服务器特性在下列情况下会出现消息生产者和消息消费者不一致情况:

1) 消费者成功从消息服务器消费消息, 当发送消费成功回执给消息服务器时, 网络出现异常, 服务器没有收到回执, 则这条消息消费者还可以再消费, 会导致消费者重复消费消息现象。

2) 当生产者成功创建消息时, 网络出现异常, 导致消息没有发送到消息服务器, 会导致消息丢失现象。

3) 当消息生产者的生产消息过程和对应的业务不在一个事务中, 相对生产者业务会出现产生多余消息或者丢失消息现象。

4) 当消息消费者的消费消息过程和对应的业务不在一个事务中, 相对消费业务会也会出现消息丢失或者消息重复消费现象。

5) 当消息服务器出现异常时, 会出现消息丢失现象。

从上述现象中, 可以总结出三类问题, 本地事务和消息不一致、网络和服务器异常、消费者重复消费。可以通过如下的机制解决上述问题, 通过消息消费者的幂等性, 解决消息重复消费问题, 具体实现方式是:

1) 通过为每个业务类型的业务单据设计唯一编号并通过该编号关联消费记录，实现对消息消费的感知。当消费者消费消息时, 先比较该消息所对应的业务编号是否消费过, 如果消费过则不触发业务但向消息服务器返回消费成功标识, 如果没有消费过则消费该消息。

2) 通过本地事务解决生产者和消费者的业务与消息的一致性问题。具体实现就是当对应业务成功时, 在本地数据库中记录消息的生产和消费。

3) 通过一个后台监控机制实现网络异常和消息服务器异常导致的消息不一致问题。具体实现就是通过比较数据库的记录, 发现不一致消息, 并对不一致消息作相应的处理。

1.3 一致性体保障平台系结构设计 1.3.1 总体体系结构设计

从基于消息通信的分布式系统看, 一致性保障层处于业务应用系统和消息中间件之间, 实现业务系统消息通信的最终一致性；从一致性实现机制看, 一致性体保障层的系结构主要组成:消息生产者、消息消费者、消息监控、消息幂等处理、消息查询、保障层后台管理, 见图 3。

消息生产者记录和业务一致的消息生产记录, 消息消费者记录和业务一致的消费记录, 保障层后台管理用于配置数据源、消息监视规则、控制规则和其他系统配置, 消息查询用于查询消息生产消费状态。

消息幂等处理:为了提升系统性能, 通过分布式缓存服务器记录每个应用已经消费的消息对应的业务唯一编号, 设置15天前的消息持久化到硬盘, 保证有足够的物理内存存储这些唯一业务编号, 当有业务编号请求, 如果在缓存中能查到则不用再消费, 如果没有查到, 则消费消息, 并保存消息。采用缓存服务器记录业务编号的好处有两点：一是性能提升；二是把比较功能从业务代码中分离到框架层。

消息监控:在后台启动一个监控进程, 遍历消息生产消费记录, 按照监控规则, 如果在一定时间消费端没有出现已经生产的消息, 则可能是消息丢失或者消息延迟, 按照控制规则, 可以再生成相同的消息让消费者消费。

更复杂的监控情况是, 当分布式交互的各方如果有一个执行失败, 其他的各方都要停止执行和回滚已经执行的结果, 消费者接收到消息, 发起业务处理, 当业务处理失败时, 消费者把业务处理失败状态写入到对应的消息, 当监控到消费者的业务处理状态是失败时, 按照监控规则发起对应的补偿操作, 撤销消息生产者的业务操作和其他消费者的业务操作, 即进行事务的补偿操作。

1.3.2 消息监控模块体系结构设计

消息监控部件是整个一致性保障平台的核心, 包含了复杂的逻辑, 从软件设计目标可扩展、可复用和易读方面来看, 采用多维视角的横切思想, 并且采用正交的软件体系结构对消息监控进行设计, 如图 4。系统初始化是在系统启动时根据系统配置启动对应的消息监控进程; 消息监控是监控进程读取到消息生产和消费记录及其状态, 再根据每种消息的监控规则进行确认完成、消息重发和消息补偿。

2 一致性保障层关键功能设计和实现

根据前面的分析关键点有:通过本地事务实现生产者和消费者保障消息记录和业务一致性, 消费者幂等处理, 消息监控机制。本章将从以上几点分别进行描述。

2.1 生产者创建消息设计实现

分布式系统在很多情况下都会出现不一致，而且也无法避免这些情况的发生，但是如果能把所有通信和事件都记录下来，那么就能感知到系统出现不一致，所以通过本地事务，在事务中同时记录业务发生数据和消息发送备案数据，就能保证业务数据和消息发送备案数据强一致。

通过图 5, 可以知道消息生产消费过程的一个概貌, 这些实体之间的关系, 通过本地事物实现业务和消息数据一致, 如果消息发送失败, 在后续的消息监控中可以发现并进行消息重发, 确保该消息一定发送成功。

代码注意要点是两个保存 (保存业务信息与保存消息内容) 在一个事务, 而且发送消息必须有异常处理, 并且由单独的线程处理, 这样即使发送消息出现网络延迟或者异常都不会影响业务实现, 没有发送成功的消息平台的监控进程会在后台主动重试发送, 具体实现代码如下:

public void bizService (){

  saveBizData ();                       //保存业务数据

  saveMegData ();                       //保存消息数据

try {

    SendMegThread sendMegThread=new SendMegThread ();

    sendMegThread.start ();             //新起线程发送消息

  } catch (Exception e) {

                       // TODO Auto-generated catch block

        e.printStackTrace ();

  }

}

2.2 消息监控设计实现

消息监控是一致性保障的核心模块，该模块包含了监视规则和控制规则。首先启动两个进程分别监控生产端和消费端, 然后按照规则, 如果匹配成功则标识为监控结束, 如果出现异常则发送补偿消息, 如果丢失或者延迟则发送重试消息。主要监控流程如图 6, 为了简单明了图中省略了多个消费者。

采用spring3定时框架启动后台监控进程, 具体配置:

< context:component-scan base-package="com.***.timer" />

< !--Enables the Spring Task @Scheduled programming model-->

< task:executor id="executor" pool-size="1" />

< task:scheduler id="scheduler" pool-size="1" />

< task:annotation-driven execu-tor="executor" scheduler=

"scheduler" />

在定时任务线程中调用监控功能, 功能接口如下:

public List < ConsumerMeg> queryConsumer (); //查询待匹配消费消息

  public List < ProducerMeg> queryProduc-er (); //查询待匹配生产者消息

  public void matchMeg (List < ConsumerMeg> listConsumerMeg, List < ProducerMeg> listProduc-erMeg); //匹配生产者和消费者消息

  public void sendRetryMeg (ProducerMeg producerMeg); //发送重试

  public void sendEqualizeMeg (ConsumerMeg consumerMeg); //发送补偿

2.3 幂等处理设计和实现

由于消息通信方式, 消息传递也是通过网络实现, 如果出现网络异常需要容错机制, 消费者从消息服务器获取消息, 并成功消费, 当反馈成功状态给消息服务器时出现网络异常, 则消息服务器还会继续保留此消息, 消费者还能继续消费该消息, 为了保证消费者不重复消费, 采用缓存服务器作为幂等^[20]处理机制。具体实现采用redis作为缓存集群服务器, 由集群层框架提供一致性性Hash和负载均衡策略, 在客户端通过jedis框架访问缓存, 系统初始化时创建连接池, 提升系统性能, 采用map数据结构存储唯一业务编码, 代码中需要注意使用连接池, 并且使用完需立即返回连接池。其主要代码如下:

public static boolean existsKey (String key){

  Boolean value= false;

  JedisPool pool=null;

  Jedis jedis=null;

  try {

    pool=getPool ();

    jedis= pool.getResource ();

    value= jedis. exists (key);

  } catch (Exception e) {                //释放redis对象

    pool.returnBrokenResource (jedis);

    e.printStackTrace ();

  } finally {                //返还到连接池

      returnResource (pool, jedis);

  }

  return value;

}

2.4 数据设计 2.4.1 tag设计

在消息中间件中消息由消息主题topic、消息标签tag和消息内容组成, topic规定消息在消息中间件中的哪个队列存储, 消息内容与该条消息关联的业务相关, 这两个单元的内容都是客观的, 不能随意改变。通过设计tag的组成, 可以赋予消息更多的内涵, 使消息中间件同时为多个业务使用, 并且保证每条消息唯一性和时效性, 具体见表 1。数据对象为tag。

2.4.2 数据库设计

数据库设计^[21]的出发点是系统要实现的功能和目标, 在前面已经详细分析了系统的目标和要实现的功能, 根据数据库设计的方法, 确定数据实体有:生产者消息实体、消费者消息实体、督查结果消息实体和配置所包含的一组实体。

1) 生产者消费者消息实体。

表示哪个业务系统的哪个节点什么时间产生了一条什么样的消息, 或者哪个系统的节点消费了一条消息, 以及该消息什么时候被某个对象消费, 为了节省存储空间, 没有记录消息内容。见表 1, 生产者内容由九个属性组成 (即表 1中第一列的内容为tag和生产者的九行), 消费者的内容也由九个属性组成 (即表 1中第一列的内容为tag和消费者的九行)。

2) 后台检查结果。

后台进程获取上面的生产者和消费者数据, 通过全局唯一标识关联数据, 再通过匹配规则, 可以得到如下结果:生产消费正常; 生产和消费不一致但还没有触发异常处理; 消费延迟或者消息丢失, 需要重试发起消息; 消费者处理出现异常, 对应的操作需要回滚。这些操作和状态通过表 1中tag和检查数据对象的属性集合记录下来。

业务系统编号:用于标识每条消息是属于哪个业务系统, 由封装的客户端从配置文件获取, 对使用消息中间件的业务系统透明。

主机编号:标识这条消息是集群中哪台机器发出的, 由封装的客户端从配置文件获取, 对使用消息中间件的业务系统透明。

业务类型:标识属于哪个业务。

时间戳:标识这条消息的创建时间, 由消息中间件客户端在发送时产生, 对使用消息中间件的业务系统透明。

业务编号:由使用消息中间件的业务系统传送，可以通过这个内容追溯到具体业务。

消息内容:记录该消息的内容。

2.5 实验和性能分析

从前面的分析中可知, 上面的设计相比传统消息系统增加了两个记录过程, 这个记录过程可能会对性能带来损耗, 从测试来看, 在双核CPU, 内存为6 GB，操作系统是SUSI Linux, MySQL版本是5.6.15, 采用单条插入的方式, 每行只有10个字段的情况下, 每秒大约2 000条数据, 这也就是其性能极限。在实际应用中单业务系统业务量很难达到这个极限，即使达到极限也可以通过批量插入提升上限，所以对性能的损耗可以忽略。

从并发性角度再对比分布式系统采用RPC通信方式和消息通信方式对整个分布式系统性能和吞吐量的影响, 如图 7所示, 表示一个总任务分别调用三个子任务, 每个子任务又分别使用数据库资源, 协同完成总任务。

为了比较RPC通信方式和消息通信方式效率，假定一个事物，需要多个子系统协作完成的场景，需要分别表示任务、资源、任务对资源的占用、任务执行的时间，具体表示如下：

任务集合T={T₁, T₂, …, T_n};

资源集合R={R₁, R₂, …, R_m}。

任务使用的资源关系表示每个任务可能使用的任意个资源：

TR{{R₁, R₂, …, R_m}, …, {R₁, R₂, …, R_n}}

一个进程需要引用的子任务Process{T₁T₂…T_k}(k < n)。

下面分别计算两种通信方式完成一个总任务需要的时间和资源的最大吞吐量。整个服务完成时间=计算时间+I/O时间+网络时间+资源等待时间。为了方便计算, 把和通信效率无关的耗时都忽略掉, 例如把占时比例极小的计算时间和I/O时间忽略, 同时考虑到可能发生的资源冲突导致等待时间，如果在RPC同步通信方式下该时间会被累计，而在消息通信方式下不会累计，即该因素对RPC通信方式的影响更大，为了比较它们优劣也可以忽略掉, 仅仅保留网络通信时间。假设每次网络通信时间为time：

RPC通信方式任务完成时间为PT_RPC=2×P_m×time(P_m为协调的子任务，每个任务耗时2×time，即请求时间+应答时间);

消息通信方式时间为PT_MSG=time+2×time。

在并发场景中资源的限制往往成了系统的瓶颈，对资源占用的耗时越小，系统的吞吐量就越大，P₁表示一个具体事务，对应的子任务集合{T₁，_T2, …, T_m}，这些子任务使用的资源集合R₁是{R₁，R₂, …，R_n}, 在一个事物中都是开始事务时分配所有资源，直到事务结束才收回资源，结合这个原理，再分析一个事务在不同通信方式下的资源占用时间, RPC通信方式资源集合R₁的占用时间是任务执行时间是PT_RPC消息通信方式资源占用时间是PT_MSG

从上面分析可知, RPC通信方式任务执行时间随着参与方增加而增加, 而消息通信方式不会; 对资源占用时间，RPC方式会随着参与方增加而增加, 但消息通信方式不会。

通过以上分析可知消息通信的效率更高，而且资源占用时间更短，在高并发场景下不易发生资源冲突，在实际运行中服务质量度量是对多个指标进行建模^[22], 不同进程之间是相互影响的一个动态过程, 例如本例中, 由于运行时间加长会导致资源锁定时间增加, 资源锁定时间增加会降低系统吞吐量, 从而降低了系统提供服务的能力。

3 实例应用

该平台和其他相关系统部署见图 8, 包括数据服务器、消息集群服务器、缓存集群服务器等, 在这种部署下为实现整个平台高可用和高性能提供基础设施保障。

在实际应用中, 消息中间件服务器采用双核CPU、8 GB内存, 操作系统是Centos 6.5, 消息中间件同时为风控系统、财务系统、网站系统、移动系统提供消息服务。图 9为对应系统在10 min内各时刻的消息吞吐量。从图 9中可以看到, 按照每分钟采集各系统的消息处理量, 最上面的曲线反映网站系统消息吞吐量, 约每分钟2 000条, 财务系统吞吐量最小。

消息督查是通过定时任务发现消息丢失和消息重复, 见图 10, 该图描述了每条消息在生产端、消息中间件和消费端的状态, 从状态可知该消息是否发生异常。对于出现异常的消息可以人工处理或者通过补偿措施自动处理, 从而保证系统最后一致。例如客户注册成功, 当发送通知消息和发送优惠券重复, 根据系统设置, 出现异常消息在3 s之内即通知, 这时如果通知消息没有发送则可以取消发送, 赠送的优惠券可以冲红处理; 当注册成功, 但是消息中间件没有收到, 那么根据系统设置30 s即报告异常, 这时通过人工处理或者补偿操作即可保持系统一致。

在实际应用中一致性保障平台在保障分布式系统一致性方面发挥了巨大作用, 表 2中的状态补偿指逆向操作, 从数据上看, 每天发生的每一次通信都能被平台捕获, 并且对其状态进行跟踪, 即时纠正了不一致数据。如果未采用这个平台, 数据不一致发生后需要在日终对账后才能修正, 可见一致性保障平台是分布式交易系统的核心组件。

从应用来看, 通过缓存系统和异步消息通信提升系统性能, 通过部署高可用集群提供高可用性, 通过一致性保障平台提供一致性。

4 结语

本文设计实现了在分布式交易环境中保障系统最终一致性的平台, 该平台通过本地事务、补偿机制和幂等性, 在满足系统性能的前提下实现可以保障在分布式系统中事务的最终一致性。该平台实现过程中考虑了软件工程化思想采用复用思想抽象为可扩展的平台, 提升了平台的经济价值。通过实验和分析可知该平台具有高性能, 并且采用高可用集群, 从实际应用来看, 为业务软件快速开发提供技术保障，在实际应用中发挥了强大的作用, 创造了很大的价值。