Hmily：高性能异步分布式事务TCC框架详解架构师

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(); 
    } 
}

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

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; 
    } 
}

/** 
 * 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; 
}

/** 
 * 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); 
    }); 
}

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); 
        } 
    } 
 
}

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

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; 
 
}