-
Notifications
You must be signed in to change notification settings - Fork 253
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request !82 from 扫地猿/master
- Loading branch information
Showing
11 changed files
with
651 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
242 changes: 242 additions & 0 deletions
242
src/main/java/io/jboot/support/seata/tcc/JbootTCCResourceManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,242 @@ | ||
| package io.jboot.support.seata.tcc; | ||
|
|
||
| import com.alibaba.fastjson.JSON; | ||
| import io.seata.common.Constants; | ||
| import io.seata.common.exception.ShouldNeverHappenException; | ||
| import io.seata.common.exception.SkipCallbackWrapperException; | ||
| import io.seata.common.util.StringUtils; | ||
| import io.seata.core.exception.TransactionException; | ||
| import io.seata.core.model.BranchStatus; | ||
| import io.seata.core.model.BranchType; | ||
| import io.seata.core.model.Resource; | ||
| import io.seata.rm.AbstractResourceManager; | ||
| import io.seata.rm.tcc.TCCResource; | ||
| import io.seata.rm.tcc.TwoPhaseResult; | ||
| import io.seata.rm.tcc.api.BusinessActionContext; | ||
|
|
||
| import java.lang.reflect.Method; | ||
| import java.lang.reflect.UndeclaredThrowableException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * @author zhangxn | ||
| */ | ||
| public class JbootTCCResourceManager extends AbstractResourceManager { | ||
|
|
||
| /** | ||
| * TCC resource cache | ||
| */ | ||
| private Map<String, Resource> tccResourceCache = new ConcurrentHashMap<>(); | ||
|
|
||
| /** | ||
| * Instantiates a new Tcc resource manager. | ||
| */ | ||
| public JbootTCCResourceManager() { | ||
| // not do anything | ||
| } | ||
|
|
||
| /** | ||
| * registry TCC resource | ||
| * | ||
| * @param resource The resource to be managed. | ||
| */ | ||
| @Override | ||
| public void registerResource(Resource resource) { | ||
| TCCResource tccResource = (TCCResource)resource; | ||
| tccResourceCache.put(tccResource.getResourceId(), tccResource); | ||
| super.registerResource(tccResource); | ||
| } | ||
|
|
||
| @Override | ||
| public Map<String, Resource> getManagedResources() { | ||
| return tccResourceCache; | ||
| } | ||
|
|
||
| /** | ||
| * TCC branch commit | ||
| * | ||
| * @param branchType | ||
| * @param xid Transaction id. | ||
| * @param branchId Branch id. | ||
| * @param resourceId Resource id. | ||
| * @param applicationData Application data bind with this branch. | ||
| * @return BranchStatus | ||
| * @throws TransactionException TransactionException | ||
| */ | ||
| @Override | ||
| public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, | ||
| String applicationData) throws TransactionException { | ||
| TCCResource tccResource = (TCCResource)tccResourceCache.get(resourceId); | ||
| if (tccResource == null) { | ||
| throw new ShouldNeverHappenException(String.format("TCC resource is not exist, resourceId: %s", resourceId)); | ||
| } | ||
| Object targetTCCBean = tccResource.getTargetBean(); | ||
| Method commitMethod = tccResource.getCommitMethod(); | ||
| if (targetTCCBean == null || commitMethod == null) { | ||
| throw new ShouldNeverHappenException(String.format("TCC resource is not available, resourceId: %s", resourceId)); | ||
| } | ||
| try { | ||
| //BusinessActionContext | ||
| BusinessActionContext businessActionContext = getBusinessActionContext(xid, branchId, resourceId, | ||
| applicationData); | ||
| Object[] args = this.getTwoPhaseCommitArgs(tccResource, businessActionContext); | ||
| Object ret; | ||
| boolean result; | ||
| // add idempotent and anti hanging | ||
| if (Boolean.TRUE.equals(businessActionContext.getActionContext(Constants.USE_TCC_FENCE))) { | ||
| try { | ||
| result = TCCFenceHandler.commitFence(commitMethod, targetTCCBean, xid, branchId, args); | ||
| } catch (SkipCallbackWrapperException | UndeclaredThrowableException e) { | ||
| throw e.getCause(); | ||
| } | ||
| } else { | ||
| ret = commitMethod.invoke(targetTCCBean, null); | ||
| if (ret != null) { | ||
| if (ret instanceof TwoPhaseResult) { | ||
| result = ((TwoPhaseResult)ret).isSuccess(); | ||
| } else { | ||
| result = (boolean)ret; | ||
| } | ||
| } else { | ||
| result = true; | ||
| } | ||
| } | ||
| LOGGER.info("TCC resource commit result : {}, xid: {}, branchId: {}, resourceId: {}", result, xid, branchId, resourceId); | ||
| return result ? BranchStatus.PhaseTwo_Committed : BranchStatus.PhaseTwo_CommitFailed_Retryable; | ||
| } catch (Throwable t) { | ||
| String msg = String.format("commit TCC resource error, resourceId: %s, xid: %s.", resourceId, xid); | ||
| LOGGER.error(msg, t); | ||
| return BranchStatus.PhaseTwo_CommitFailed_Retryable; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * TCC branch rollback | ||
| * | ||
| * @param branchType the branch type | ||
| * @param xid Transaction id. | ||
| * @param branchId Branch id. | ||
| * @param resourceId Resource id. | ||
| * @param applicationData Application data bind with this branch. | ||
| * @return BranchStatus | ||
| * @throws TransactionException TransactionException | ||
| */ | ||
| @Override | ||
| public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, | ||
| String applicationData) throws TransactionException { | ||
| TCCResource tccResource = (TCCResource)tccResourceCache.get(resourceId); | ||
| if (tccResource == null) { | ||
| throw new ShouldNeverHappenException(String.format("TCC resource is not exist, resourceId: %s", resourceId)); | ||
| } | ||
| Object targetTCCBean = tccResource.getTargetBean(); | ||
| Method rollbackMethod = tccResource.getRollbackMethod(); | ||
| if (targetTCCBean == null || rollbackMethod == null) { | ||
| throw new ShouldNeverHappenException(String.format("TCC resource is not available, resourceId: %s", resourceId)); | ||
| } | ||
| try { | ||
| //BusinessActionContext | ||
| BusinessActionContext businessActionContext = getBusinessActionContext(xid, branchId, resourceId, | ||
| applicationData); | ||
| Object[] args = this.getTwoPhaseRollbackArgs(tccResource, businessActionContext); | ||
|
|
||
| Object ret; | ||
| boolean result; | ||
| // add idempotent and anti hanging | ||
| if (Boolean.TRUE.equals(businessActionContext.getActionContext(Constants.USE_TCC_FENCE))) { | ||
| try { | ||
| result = TCCFenceHandler.rollbackFence(rollbackMethod, targetTCCBean, xid, branchId, | ||
| args, tccResource.getActionName()); | ||
| } catch (SkipCallbackWrapperException | UndeclaredThrowableException e) { | ||
| throw e.getCause(); | ||
| } | ||
| } else { | ||
| ret = rollbackMethod.invoke(targetTCCBean,args); | ||
| if (ret != null) { | ||
| if (ret instanceof TwoPhaseResult) { | ||
| result = ((TwoPhaseResult)ret).isSuccess(); | ||
| } else { | ||
| result = (boolean)ret; | ||
| } | ||
| } else { | ||
| result = true; | ||
| } | ||
| } | ||
| LOGGER.info("TCC resource rollback result : {}, xid: {}, branchId: {}, resourceId: {}", result, xid, branchId, resourceId); | ||
| return result ? BranchStatus.PhaseTwo_Rollbacked : BranchStatus.PhaseTwo_RollbackFailed_Retryable; | ||
| } catch (Throwable t) { | ||
| String msg = String.format("rollback TCC resource error, resourceId: %s, xid: %s.", resourceId, xid); | ||
| LOGGER.error(msg, t); | ||
| return BranchStatus.PhaseTwo_RollbackFailed_Retryable; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * transfer tcc applicationData to BusinessActionContext | ||
| * | ||
| * @param xid the xid | ||
| * @param branchId the branch id | ||
| * @param resourceId the resource id | ||
| * @param applicationData the application data | ||
| * @return business action context | ||
| */ | ||
| protected BusinessActionContext getBusinessActionContext(String xid, long branchId, String resourceId, | ||
| String applicationData) { | ||
| Map actionContextMap = null; | ||
| if (StringUtils.isNotBlank(applicationData)) { | ||
| Map tccContext = JSON.parseObject(applicationData, Map.class); | ||
| actionContextMap = (Map)tccContext.get(Constants.TCC_ACTION_CONTEXT); | ||
| } | ||
| if (actionContextMap == null) { | ||
| actionContextMap = new HashMap<>(2); | ||
| } | ||
|
|
||
| //instance the action context | ||
| BusinessActionContext businessActionContext = new BusinessActionContext( | ||
| xid, String.valueOf(branchId), actionContextMap); | ||
| businessActionContext.setActionName(resourceId); | ||
| return businessActionContext; | ||
| } | ||
|
|
||
| /** | ||
| * get phase two commit method's args | ||
| * @param tccResource tccResource | ||
| * @param businessActionContext businessActionContext | ||
| * @return args | ||
| */ | ||
| private Object[] getTwoPhaseCommitArgs(TCCResource tccResource, BusinessActionContext businessActionContext) { | ||
| String[] keys = tccResource.getPhaseTwoCommitKeys(); | ||
| Class<?>[] argsCommitClasses = tccResource.getCommitArgsClasses(); | ||
| return this.getTwoPhaseMethodParams(keys, argsCommitClasses, businessActionContext); | ||
| } | ||
|
|
||
| /** | ||
| * get phase two rollback method's args | ||
| * @param tccResource tccResource | ||
| * @param businessActionContext businessActionContext | ||
| * @return args | ||
| */ | ||
| private Object[] getTwoPhaseRollbackArgs(TCCResource tccResource, BusinessActionContext businessActionContext) { | ||
| String[] keys = tccResource.getPhaseTwoRollbackKeys(); | ||
| Class<?>[] argsRollbackClasses = tccResource.getRollbackArgsClasses(); | ||
| return this.getTwoPhaseMethodParams(keys, argsRollbackClasses, businessActionContext); | ||
| } | ||
|
|
||
| private Object[] getTwoPhaseMethodParams(String[] keys, Class<?>[] argsClasses, BusinessActionContext businessActionContext) { | ||
| Object[] args = new Object[argsClasses.length]; | ||
| for (int i = 0; i < argsClasses.length; i++) { | ||
| if (argsClasses[i].equals(BusinessActionContext.class)) { | ||
| args[i] = businessActionContext; | ||
| } else { | ||
| args[i] = businessActionContext.getActionContext(keys[i], argsClasses[i]); | ||
| } | ||
| } | ||
| return args; | ||
| } | ||
|
|
||
| @Override | ||
| public BranchType getBranchType() { | ||
| return BranchType.TCC; | ||
| } | ||
| } |
Oops, something went wrong.