调整优化

Author: xlorne

tx-manager/src/main/java/com/codingapi/tm/compensate/service/impl/CompensateServiceImpl.java

@@ -57,68 +57,64 @@ public boolean saveCompensateMsg(TransactionCompensateMsg transactionCompensateM
         String key = configReader.getKeyPrefix() + transactionCompensateMsg.getGroupId();
         TxGroup txGroup = redisServerService.getTxGroupByKey(key);
         if (txGroup == null) {
-//            key = configReader.getKeyPrefixNotify() + transactionCompensateMsg.getGroupId();
-//            txGroup = redisServerService.getTxGroupByKey(key);
-            //todo  待完善
+            return false;
         }
-        if(txGroup!=null) {
-            redisServerService.deleteKey(key);
-
-            //已经全部通知的模块不做补偿处理
-            boolean hasNoNotify = false;
-            for(TxInfo txInfo:txGroup.getList()){
-                if(txInfo.getNotify()==0){
-                    hasNoNotify = true;
-                }
-            }
 
-            if(!hasNoNotify){
-                //事务已经执行完毕的
-                return true;
-            }
+        redisServerService.deleteKey(key);
 
+        //已经全部通知的模块不做补偿处理
+        boolean hasNoNotify = false;
 
+        for(TxInfo txInfo:txGroup.getList()){
+            if(txInfo.getNotify()==0){
+                hasNoNotify = true;
+            }
+        }
+
+        if(!hasNoNotify){
+            //事务已经执行完毕的
+            logger.info("TxGroup had notify ! ");
+            return true;
+        }
 
-            transactionCompensateMsg.setTxGroup(txGroup);
 
-            final String json = JSON.toJSONString(transactionCompensateMsg);
+        transactionCompensateMsg.setTxGroup(txGroup);
 
-            logger.info("补偿->" + json);
+        final String json = JSON.toJSONString(transactionCompensateMsg);
 
-            final String compensateKey = compensateDao.saveCompensateMsg(transactionCompensateMsg);
+        logger.info("补偿->" + json);
 
-            //调整自动补偿机制，若开启了自动补偿，需要通知业务返回success，方可执行自动补偿
-            threadPool.execute(new Runnable() {
-                @Override
-                public void run() {
-                    try {
-                        String groupId = transactionCompensateMsg.getGroupId();
-                        JSONObject requestJson = new JSONObject();
-                        requestJson.put("action", "compensate");
-                        requestJson.put("groupId", groupId);
-                        requestJson.put("json", json);
+        final String compensateKey = compensateDao.saveCompensateMsg(transactionCompensateMsg);
 
-                        String url = configReader.getCompensateNotifyUrl();
-                        logger.error("补偿回调地址->" + url);
-                        String res = HttpUtils.postJson(url, requestJson.toJSONString());
-                        logger.error("补偿回调结果->" + res);
-                        if (configReader.isCompensateAuto()) {
-                            //自动补偿,是否自动执行补偿
-                            if (res.contains("success")||res.contains("SUCCESS")) {
-                                //自动补偿
-                                autoCompensate(compensateKey, transactionCompensateMsg);
-                            }
+        //调整自动补偿机制，若开启了自动补偿，需要通知业务返回success，方可执行自动补偿
+        threadPool.execute(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    String groupId = transactionCompensateMsg.getGroupId();
+                    JSONObject requestJson = new JSONObject();
+                    requestJson.put("action", "compensate");
+                    requestJson.put("groupId", groupId);
+                    requestJson.put("json", json);
+
+                    String url = configReader.getCompensateNotifyUrl();
+                    logger.error("补偿回调地址->" + url);
+                    String res = HttpUtils.postJson(url, requestJson.toJSONString());
+                    logger.error("补偿回调结果->" + res);
+                    if (configReader.isCompensateAuto()) {
+                        //自动补偿,是否自动执行补偿
+                        if (res.contains("success")||res.contains("SUCCESS")) {
+                            //自动补偿
+                            autoCompensate(compensateKey, transactionCompensateMsg);
                         }
-                    } catch (Exception e) {
-                        logger.error("补偿回调失败->" + e.getMessage());
                     }
+                } catch (Exception e) {
+                    logger.error("补偿回调失败->" + e.getMessage());
                 }
-            });
+            }
+        });
 
-            return StringUtils.isNotEmpty(compensateKey);
-        }else {
-            return false;
-        }
+        return StringUtils.isNotEmpty(compensateKey);
 
     }
 

tx-manager/src/main/java/com/codingapi/tm/manager/service/impl/TxManagerServiceImpl.java

@@ -97,10 +97,12 @@ public int cleanNotifyTransaction(String groupId, String taskId) {
         String key = configReader.getKeyPrefix() + groupId;
         TxGroup txGroup = redisServerService.getTxGroupByKey(key);
         if (txGroup==null) {
+            logger.info("cleanNotifyTransaction - > txGroup is null ");
             return res;
         }
 
-        if(txGroup.getState()==0){
+        if(txGroup.getHasOver()==0){
+            logger.info("cleanNotifyTransaction - > groupId "+groupId+" not over !");
             return 0;
         }
 
@@ -150,6 +152,7 @@ public boolean closeTransactionGroup(String groupId,int state) {
         }
         txGroup.setState(state);
         txGroup.setHasOver(1);
+        redisServerService.saveTransaction(key,txGroup.toJsonString());
         return transactionConfirmService.confirm(txGroup);
     }
 

tx-manager/src/main/java/com/codingapi/tm/netty/service/impl/ActionCTGServiceImpl.java

@@ -2,7 +2,6 @@
 
 import com.alibaba.fastjson.JSONObject;
 import com.codingapi.tm.manager.service.TxManagerService;
-import com.codingapi.tm.netty.model.TxGroup;
 import com.codingapi.tm.netty.service.IActionService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;