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