热门搜索:
您现在的位置是：首页 > 文章详情

Hmily：高性能异步分布式事务TCC框架

日期：2018-10-11点击：362

Hmily框架特性

[https://github.com/yu199195/hmily]


  • 无缝集成Spring,Spring boot start。

  • 无缝集成Dubbo,SpringCloud,Motan等rpc框架。

  • 多种事务日志的存储方式（redis，mongdb,mysql等）。

  • 多种不同日志序列化方式（Kryo,protostuff,hession）。

  • 事务自动恢复。

  • 支持内嵌事务的依赖传递。

  • 代码零侵入,配置简单灵活。


Hmily为什么这么高性能？


1.采用disruptor进行事务日志的异步读写（disruptor是一个无锁，无GC的并发编程框架）


package com.hmily.tcc.core.disruptor.publisher;

import com.hmily.tcc.common.bean.entity.TccTransaction;
import com.hmily.tcc.common.enums.EventTypeEnum;
import com.hmily.tcc.core.concurrent.threadpool.HmilyThreadFactory;
import com.hmily.tcc.core.coordinator.CoordinatorService;
import com.hmily.tcc.core.disruptor.event.HmilyTransactionEvent;
import com.hmily.tcc.core.disruptor.factory.HmilyTransactionEventFactory;
import com.hmily.tcc.core.disruptor.handler.HmilyConsumerDataHandler;
import com.hmily.tcc.core.disruptor.translator.HmilyTransactionEventTranslator;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
  * event publisher.
  *
  * @author xiaoyu(Myth)
  */
@Component
public class HmilyTransactionEventPublisher implements DisposableBean {

     private Disruptor<HmilyTransactionEvent> disruptor;

     private final CoordinatorService coordinatorService;

     @Autowired
     public HmilyTransactionEventPublisher(final CoordinatorService coordinatorService) {
         this.coordinatorService = coordinatorService;
     }

     /**
      * disruptor start.
      *
      * @param bufferSize this is disruptor buffer size.
      * @param threadSize this is disruptor consumer thread size.
      */
     public void start(final int bufferSize, final int threadSize) {
         disruptor = new Disruptor<>(new HmilyTransactionEventFactory(), bufferSize, r -> {
             AtomicInteger index = new AtomicInteger(1);
             return new Thread(null, r, "disruptor-thread-" + index.getAndIncrement());
         }, ProducerType.MULTI, new BlockingWaitStrategy());

         final Executor executor = new ThreadPoolExecutor(threadSize, threadSize, 0, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<>(),
                 HmilyThreadFactory.create("hmily-log-disruptor", false),
                 new ThreadPoolExecutor.AbortPolicy());

         HmilyConsumerDataHandler[] consumers = new HmilyConsumerDataHandler[threadSize];
         for (int i = 0; i < threadSize; i++) {
             consumers[i] = new HmilyConsumerDataHandler(executor, coordinatorService);
         }
         disruptor.handleEventsWithWorkerPool(consumers);
         disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
         disruptor.start();
     }

     /**
      * publish disruptor event.
      *
      * @param tccTransaction {@linkplain com.hmily.tcc.common.bean.entity.TccTransaction }
      * @param type           {@linkplain EventTypeEnum}
      */
     public void publishEvent(final TccTransaction tccTransaction, final int type) {
         final RingBuffer<HmilyTransactionEvent> ringBuffer = disruptor.getRingBuffer();
         ringBuffer.publishEvent(new HmilyTransactionEventTranslator(type), tccTransaction);
     }

     @Override
     public void destroy() {
         disruptor.shutdown();
     }
 }


在这里bufferSize 的默认值是4094 * 4,用户可以根据自行的情况进行配置。


   HmilyConsumerDataHandler[] consumers = new HmilyConsumerDataHandler[threadSize];
         for (int i = 0; i < threadSize; i++) {
             consumers[i] = new HmilyConsumerDataHandler(executor, coordinatorService);
         }
         disruptor.handleEventsWithWorkerPool(consumers);


这里是采用多个消费者去处理队列里面的任务。


2.异步执行confrim,cancel方法。


package com.hmily.tcc.core.service.handler;

import com.hmily.tcc.common.bean.context.TccTransactionContext;
import com.hmily.tcc.common.bean.entity.TccTransaction;
import com.hmily.tcc.common.enums.TccActionEnum;
import com.hmily.tcc.core.concurrent.threadpool.HmilyThreadFactory;
import com.hmily.tcc.core.service.HmilyTransactionHandler;
import com.hmily.tcc.core.service.executor.HmilyTransactionExecutor;
import org.aspectj.lang.ProceedingJoinPoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
  * this is transaction starter.
  *
  * @author xiaoyu
  */
@Component
public class StarterHmilyTransactionHandler implements HmilyTransactionHandler {

     private static final int MAX_THREAD = Runtime.getRuntime().availableProcessors() << 1;

     private final HmilyTransactionExecutor hmilyTransactionExecutor;

     private final Executor executor = new ThreadPoolExecutor(MAX_THREAD, MAX_THREAD, 0, TimeUnit.MILLISECONDS,
             new LinkedBlockingQueue<>(),
             HmilyThreadFactory.create("hmily-execute", false),
             new ThreadPoolExecutor.AbortPolicy());

     @Autowired
     public StarterHmilyTransactionHandler(final HmilyTransactionExecutor hmilyTransactionExecutor) {
         this.hmilyTransactionExecutor = hmilyTransactionExecutor;
     }

     @Override
     public Object handler(final ProceedingJoinPoint point, final TccTransactionContext context)
             throws Throwable {
         Object returnValue;
         try {
             TccTransaction tccTransaction = hmilyTransactionExecutor.begin(point);
             try {
                 //execute try
                 returnValue = point.proceed();
                 tccTransaction.setStatus(TccActionEnum.TRYING.getCode());
                 hmilyTransactionExecutor.updateStatus(tccTransaction);
             } catch (Throwable throwable) {
                 //if exception ,execute cancel
                 final TccTransaction currentTransaction = hmilyTransactionExecutor.getCurrentTransaction();
                 executor.execute(() -> hmilyTransactionExecutor
                         .cancel(currentTransaction));
                 throw throwable;
             }
             //execute confirm
             final TccTransaction currentTransaction = hmilyTransactionExecutor.getCurrentTransaction();
             executor.execute(() -> hmilyTransactionExecutor.confirm(currentTransaction));
         } finally {
             hmilyTransactionExecutor.remove();
         }
         return returnValue;
     }
 }


当try方法的AOP切面有异常的时候，采用线程池异步去执行cancel，无异常的时候去执行confrim方法。


这里有人可能会问：那么cancel方法异常，或者confrim方法异常怎么办呢？

答：首先这种情况是非常罕见的，因为你上一面才刚刚执行完try。其次如果出现这种情况，在try阶段会保存好日志，Hmily有内置的调度线程池来进行恢复，不用担心。


有人又会问：这里如果日志保存异常了怎么办？


答：首先这又是一个牛角尖问题，首先日志配置的参数，在框架启动的时候，会要求你配置的。其次，就算在运行过程中日志保存异常，这时候框架会取缓存中的，并不会影响程序正确执行。最后，万一日志保存异常了，系统又在很极端的情况下down机了，恭喜你，你可以去买彩票了，最好的解决办法就是不去解决它。


3.ThreadLocal缓存的使用。


  /**
      * transaction begin.
      *
      * @param point cut point.
      * @return TccTransaction
      */
     public TccTransaction begin(final ProceedingJoinPoint point) {
         LogUtil.debug(LOGGER, () -> "......hmily transaction！start....");
         //build tccTransaction
         final TccTransaction tccTransaction = buildTccTransaction(point, TccRoleEnum.START.getCode(), null);
         //save tccTransaction in threadLocal
         CURRENT.set(tccTransaction);
         //publishEvent
         hmilyTransactionEventPublisher.publishEvent(tccTransaction, EventTypeEnum.SAVE.getCode());
         //set TccTransactionContext this context transfer remote
         TccTransactionContext context = new TccTransactionContext();
         //set action is try
         context.setAction(TccActionEnum.TRYING.getCode());
         context.setTransId(tccTransaction.getTransId());
         context.setRole(TccRoleEnum.START.getCode());
         TransactionContextLocal.getInstance().set(context);
         return tccTransaction;
     }


首先要理解，threadLocal保存的发起者一方法的事务信息。这个很重要，不要会有点懵逼。rpc的调用，会形成调用链，进行保存。


 /**
      * add participant.
      *
      * @param participant {@linkplain Participant}
      */
     public void enlistParticipant(final Participant participant) {
         if (Objects.isNull(participant)) {
             return;
         }
         Optional.ofNullable(getCurrentTransaction())
                 .ifPresent(c -> {
                     c.registerParticipant(participant);
                     updateParticipant(c);
                 });
     }

4.GuavaCache的使用


package com.hmily.tcc.core.cache;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.Weigher;
import com.hmily.tcc.common.bean.entity.TccTransaction;
import com.hmily.tcc.core.coordinator.CoordinatorService;
import com.hmily.tcc.core.helper.SpringBeanUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.Optional;
import java.util.concurrent.ExecutionException;

/**
  * use google guava cache.
  * @author xiaoyu
  */
public final class TccTransactionCacheManager {

     private static final int MAX_COUNT = 10000;

     private static final LoadingCache<String, TccTransaction> LOADING_CACHE =
             CacheBuilder.newBuilder().maximumWeight(MAX_COUNT)
                     .weigher((Weigher<String, TccTransaction>) (string, tccTransaction) -> getSize())
                     .build(new CacheLoader<String, TccTransaction>() {
                         @Override
                         public TccTransaction load(final String key) {
                             return cacheTccTransaction(key);
                         }
                     });

     private static CoordinatorService coordinatorService = SpringBeanUtils.getInstance().getBean(CoordinatorService.class);

     private static final TccTransactionCacheManager TCC_TRANSACTION_CACHE_MANAGER = new TccTransactionCacheManager();

     private TccTransactionCacheManager() {

     }

     /**
      * TccTransactionCacheManager.
      *
      * @return TccTransactionCacheManager
      */
     public static TccTransactionCacheManager getInstance() {
         return TCC_TRANSACTION_CACHE_MANAGER;
     }

     private static int getSize() {
         return (int) LOADING_CACHE.size();
     }

     private static TccTransaction cacheTccTransaction(final String key) {
         return Optional.ofNullable(coordinatorService.findByTransId(key)).orElse(new TccTransaction());
     }

     /**
      * cache tccTransaction.
      *
      * @param tccTransaction {@linkplain TccTransaction}
      */
     public void cacheTccTransaction(final TccTransaction tccTransaction) {
         LOADING_CACHE.put(tccTransaction.getTransId(), tccTransaction);
     }

     /**
      * acquire TccTransaction.
      *
      * @param key this guava key.
      * @return {@linkplain TccTransaction}
      */
     public TccTransaction getTccTransaction(final String key) {
         try {
             return LOADING_CACHE.get(key);
         } catch (ExecutionException e) {
             return new TccTransaction();
         }
     }

     /**
      * remove guava cache by key.
      * @param key guava cache key.
      */
     public void removeByKey(final String key) {
         if (StringUtils.isNotEmpty(key)) {
             LOADING_CACHE.invalidate(key);
         }
     }

 }


在参与者中，我们使用了ThreadLocal，而在参与者中，我们为什么不使用呢？


其实原因有二点：首先.因为try，和confrim 会不在一个线程里，会造成ThreadLocal失效。当考虑到RPC集群的时候，可能会负载到不同的机器上。


这里有一个细节就是：

   private static TccTransaction cacheTccTransaction(final String key) {
         return Optional.ofNullable(coordinatorService.findByTransId(key)).orElse(new TccTransaction());
     }


当GuavaCache里面没有的时候，会去查询日志返回，这样就保证了对集群环境的支持。


以上4点造就了Hmily是一个异步的高性能分布式事务TCC框架的原因。


Hmily如何使用？

（https://github.com/yu199195/hmily/tree/master/hmily-tcc-demo）


首先因为之前的包命名问题，框架包并没有上传到maven中心仓库，固需要使用者自己拉取代码，编译deploy到自己的私服。


1.dubbo用户


  • 在你的Api接口项目引入

  <dependency>
           <groupId>com.hmily.tcc</groupId>
           <artifactId>hmily-tcc-annotation</artifactId>
           <version>{you version}</version>
       </dependency>


  • 在你的服务提供者项目引入

 <dependency>
             <groupId>com.hmily.tcc</groupId>
             <artifactId>hmily-tcc-dubbo</artifactId>
             <version>{you version}</version>
         </dependency>


  • 配置启动bean

<!-- Aspect 切面配置，是否开启AOP切面-->
   <aop:aspectj-autoproxy expose-proxy="true"/>
   <!--扫描框架的包-->
   <context:component-scan base-package="com.hmily.tcc.*"/>
   <!--启动类属性配置-->
    <bean id="hmilyTransactionBootstrap" class="com.hmily.tcc.core.bootstrap.HmilyTransactionBootstrap">
         <property name="serializer" value="kryo"/>
         <property name="recoverDelayTime" value="120"/>
         <property name="retryMax" value="3"/>
         <property name="scheduledDelay" value="120"/>
         <property name="scheduledThreadMax" value="4"/>
         <property name="repositorySupport" value="db"/>
         <property name="tccDbConfig">
             <bean class="com.hmily.tcc.common.config.TccDbConfig">
                 <property name="url"
                           value="jdbc:mysql://192.168.1.98:3306/tcc?useUnicode=true&amp;characterEncoding=utf8"/>
                 <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
                 <property name="username" value="root"/>
                 <property name="password" value="123456"/>
             </bean>
         </property>
     </bean>


当然配置属性很多，这里我只给出了demo，具体可以参考这个类：

package com.hmily.tcc.common.config;

import com.hmily.tcc.common.enums.RepositorySupportEnum;
import lombok.Data;

/**
  * hmily config.
  *
  * @author xiaoyu
  */
 @Data
public class TccConfig {


     /**
      * Resource suffix this parameter please fill in about is the transaction store path.
      * If it's a table store this is a table suffix, it's stored the same way.
      * If this parameter is not filled in, the applicationName of the application is retrieved by default
      */
     private String repositorySuffix;

     /**
      * log serializer.
      * {@linkplain com.hmily.tcc.common.enums.SerializeEnum}
      */
     private String serializer = "kryo";

     /**
      * scheduledPool Thread size.
      */
     private int scheduledThreadMax = Runtime.getRuntime().availableProcessors() << 1;

     /**
      * scheduledPool scheduledDelay unit SECONDS.
      */
     private int scheduledDelay = 60;

     /**
      * retry max.
      */
     private int retryMax = 3;

     /**
      * recoverDelayTime Unit seconds
      * (note that this time represents how many seconds after the local transaction was created before execution).
      */
     private int recoverDelayTime = 60;

     /**
      * Parameters when participants perform their own recovery.
      * 1.such as RPC calls time out
      * 2.such as the starter down machine
      */
     private int loadFactor = 2;

     /**
      * repositorySupport.
      * {@linkplain RepositorySupportEnum}
      */
     private String repositorySupport = "db";

     /**
      * disruptor bufferSize.
      */
     private int bufferSize = 4096 * 2 * 2;

     /**
      * this is disruptor consumerThreads.
      */
     private int consumerThreads = Runtime.getRuntime().availableProcessors() << 1;

     /**
      * db config.
      */
     private TccDbConfig tccDbConfig;

     /**
      * mongo config.
      */
     private TccMongoConfig tccMongoConfig;

     /**
      * redis config.
      */
     private TccRedisConfig tccRedisConfig;

     /**
      * zookeeper config.
      */
     private TccZookeeperConfig tccZookeeperConfig;

     /**
      * file config.
      */
     private TccFileConfig tccFileConfig;

 }

2.SpringCloud用户


  • 需要引入

     <dependency>
           <groupId>com.hmily.tcc</groupId>
           <artifactId>hmily-tcc-springcloud</artifactId>
           <version>{you version}</version>
       </dependency>

  • 配置启动bean 如上。


3.Motan用户


  • 需要引入

     <dependency>
           <groupId>com.hmily.tcc</groupId>
           <artifactId>hmily-tcc-motan</artifactId>
           <version>{you version}</version>
       </dependency>


  • 配置启动bean 如上。


hmily-spring-boot-start


  • 那这个就更容易了，只需要根据你的RPC框架去引入不同的jar包。

  • 如果你是dubbo用户，那么引入

<dependency>
      <groupId>com.hmily.tcc</groupId>
      <artifactId>hmily-tcc-spring-boot-starter-dubbo</artifactId>
      <version>${your version}</version>
  </dependency>

  • 如果你是SpringCloud用户，那么引入

<dependency>
      <groupId>com.hmily.tcc</groupId>
      <artifactId>hmily-tcc-spring-boot-starter-springcloud</artifactId>
      <version>${your version}</version>
  </dependency>

  • 如果你是Motan用户，那么引入

<dependency>
      <groupId>com.hmily.tcc</groupId>
      <artifactId>hmily-tcc-spring-boot-starter-motan</artifactId>
      <version>${your version}</version>
  </dependency>

  • 然后在你的yml里面进行如下配置：

hmily:
     tcc :
         serializer : kryo
         recoverDelayTime : 128
         retryMax : 3
         scheduledDelay : 128
         scheduledThreadMax :  10
         repositorySupport : db
         tccDbConfig :
                  driverClassName  : com.mysql.jdbc.Driver
                  url :  jdbc:mysql://192.168.1.98:3306/tcc?useUnicode=true&amp;characterEncoding=utf8
                  username : root
                  password : 123456

         #repositorySupport : redis
         #tccRedisConfig:
                  #masterName: mymaster
                  #sentinel : true
                  #sentinelUrl : 192.168.1.91:26379;192.168.1.92:26379;192.168.1.93:26379
                  #password  : foobaredbbexONE123


        # repositorySupport : zookeeper
        #         host      : 92.168.1.73:2181
        #         sessionTimeOut      :  100000
        #         rootPath  : /tcc

        # repositorySupport : mongodb
        #       mongoDbUrl  : 192.168.1.68:27017
        #       mongoDbName  :  happylife
        #       mongoUserName  : xiaoyu
        #       mongoUserPwd   : 123456

        # repositorySupport : file
        #         path      : /account
        #         prefix    :  account


就这么简单，然后就可以在接口方法上加上@Tcc注解，进行愉快的使用了。

当然因为篇幅问题，很多东西只是简单的描述，尤其是逻辑方面的。


下面是github地址：https://github.com/yu199195/hmily
最后再次感谢大家，如果有兴趣的朋友，可以提供你的优秀牛逼轰轰的PR。



原文发布时间为：2018-10-12
本文作者：xiaoyu
本文来自云栖社区合作伙伴“程序猿DD”，了解相关信息可以关注“程序猿DD”。
原文链接：https://yq.aliyun.com/articles/652113
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有，本站原创内容转载请注明来源。

相关文章

    文章评论

    共有0条评论来说两句吧...

    文章二维码

    扫描即可查看该文章

    点击排行

    推荐阅读

    最新文章