删除notify格式的redis数据，保存事务源数据的一致性.减少对redis的操作

xlorne · xlorne · commit a43316f4674d · 2017-11-28T12:38:26.000+08:00
diff --git a/tx-client/src/main/java/com/codingapi/tx/datasource/service/impl/DataSourceServiceImpl.java b/tx-client/src/main/java/com/codingapi/tx/datasource/service/impl/DataSourceServiceImpl.java
@@ -1,8 +1,8 @@
 package com.codingapi.tx.datasource.service.impl;
 
-import com.lorne.core.framework.utils.task.Task;
 import com.codingapi.tx.datasource.service.DataSourceService;
 import com.codingapi.tx.netty.service.MQTxManagerService;
+import com.lorne.core.framework.utils.task.Task;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -19,26 +19,22 @@ public class DataSourceServiceImpl implements DataSourceService {
 
     @Override
     public void schedule(String groupId, Task waitTask) {
+
+
         String waitTaskId = waitTask.getKey();
-        int rs = txManagerService.checkTransactionInfo(groupId, waitTaskId);
+        int rs = txManagerService.cleanNotifyTransaction(groupId, waitTaskId);
         if (rs == 1 || rs == 0) {
             waitTask.setState(rs);
             waitTask.signalTask();
-            //clear
-            txManagerService.clearTransaction(groupId,waitTaskId,true);
 
             return;
         }
-        if(rs==-2) {
-            rs = txManagerService.getTransaction(groupId, waitTaskId);
-            if (rs == 1 || rs == 0) {
-                waitTask.setState(rs);
-                waitTask.signalTask();
-                //clear
-
-                txManagerService.clearTransaction(groupId, waitTaskId, false);
-                return;
-            }
+        rs = txManagerService.cleanNotifyTransactionHttp(groupId, waitTaskId);
+        if (rs == 1 || rs == 0) {
+            waitTask.setState(rs);
+            waitTask.signalTask();
+
+            return;
         }
 
         //添加到补偿队列
diff --git a/tx-client/src/main/java/com/codingapi/tx/netty/service/MQTxManagerService.java b/tx-client/src/main/java/com/codingapi/tx/netty/service/MQTxManagerService.java
@@ -50,26 +50,25 @@ public interface MQTxManagerService {
      * @param taskId    任务id
      * @return  事务状态
      */
-    int checkTransactionInfo(String groupId, String taskId);
+    int cleanNotifyTransaction(String groupId, String taskId);
 
 
-    /**
-     * 检查事务状态   通过http请求
-     * @param groupId 事务组id
-     * @param waitTaskId 任务id
-     * @return 事务状态
-     */
-    int getTransaction(String groupId, String waitTaskId);
+//    /**
+//     * 检查事务状态   通过http请求
+//     * @param groupId 事务组id
+//     * @param waitTaskId 任务id
+//     * @return 事务状态
+//     */
+//    int getTransaction(String groupId, String waitTaskId);
 
 
     /**
      * 检查并清理事务数据
      * @param groupId   事务组id
      * @param waitTaskId    任务id
-     * @param isGroup   是否合并事务
      * @return  事务状态
      */
-    int clearTransaction(String groupId, String waitTaskId, boolean isGroup);
+    int cleanNotifyTransactionHttp(String groupId, String waitTaskId);
 
     /**
      * 记录补偿事务数据到tm
diff --git a/tx-client/src/main/java/com/codingapi/tx/netty/service/impl/MQTxManagerServiceImpl.java b/tx-client/src/main/java/com/codingapi/tx/netty/service/impl/MQTxManagerServiceImpl.java
@@ -88,7 +88,7 @@ public void uploadModelInfo() {
     }
 
     @Override
-    public int checkTransactionInfo(String groupId, String taskId) {
+    public int cleanNotifyTransaction(String groupId, String taskId) {
         JSONObject jsonObject = new JSONObject();
         jsonObject.put("g", groupId);
         jsonObject.put("t", taskId);
@@ -101,26 +101,26 @@ public int checkTransactionInfo(String groupId, String taskId) {
         }
     }
 
-
-    @Override
-    public int getTransaction(String groupId, String waitTaskId) {
-
-        String json = managerHelper.httpGet(configReader.getTxUrl() + "getTransaction?groupId=" + groupId + "&taskId=" + waitTaskId);
-        if (json == null) {
-            return -2;
-        }
-        json = json.trim();
-        try {
-            return Integer.parseInt(json);
-        }catch (Exception e){
-            return -2;
-        }
-    }
+//
+//    @Override
+//    public int getTransaction(String groupId, String waitTaskId) {
+//
+//        String json = managerHelper.httpGet(configReader.getTxUrl() + "getTransaction?groupId=" + groupId + "&taskId=" + waitTaskId);
+//        if (json == null) {
+//            return -2;
+//        }
+//        json = json.trim();
+//        try {
+//            return Integer.parseInt(json);
+//        }catch (Exception e){
+//            return -2;
+//        }
+//    }
 
 
     @Override
-    public int clearTransaction(String groupId, String waitTaskId, boolean isGroup) {
-        String url = configReader.getTxUrl() + "clearTransaction?groupId=" + groupId + "&taskId=" + waitTaskId + "&isGroup=" + (isGroup ? 1 : 0);
+    public int cleanNotifyTransactionHttp(String groupId, String waitTaskId) {
+        String url = configReader.getTxUrl() + "cleanNotifyTransactionHttp?groupId=" + groupId + "&taskId=" + waitTaskId;
         String clearRes = managerHelper.httpGet(url);
         if(clearRes==null){
             return -1;
diff --git a/tx-manager/src/main/java/com/codingapi/tm/api/controller/TxManagerController.java b/tx-manager/src/main/java/com/codingapi/tm/api/controller/TxManagerController.java
@@ -25,17 +25,12 @@ public TxServer getServer() {
     }
 
 
-    @RequestMapping("/clearTransaction")
-    public boolean clearTransaction(@RequestParam("groupId") String groupId,@RequestParam("taskId") String taskId,@RequestParam("isGroup") int isGroup) {
-        return apiTxManagerService.clearTransaction(groupId,taskId,isGroup);
+    @RequestMapping("/cleanNotifyTransaction")
+    public int cleanNotifyTransaction(@RequestParam("groupId") String groupId,@RequestParam("taskId") String taskId) {
+        return apiTxManagerService.cleanNotifyTransaction(groupId,taskId);
     }
 
 
-    @RequestMapping("/getTransaction")
-    public int getTransaction(@RequestParam("groupId") String groupId,@RequestParam("taskId") String taskId) {
-        return apiTxManagerService.getTransaction(groupId,taskId);
-    }
-
     @RequestMapping("/sendMsg")
     public String sendMsg(@RequestParam("msg") String msg,@RequestParam("model") String model) {
         return apiTxManagerService.sendMsg(model,msg);
diff --git a/tx-manager/src/main/java/com/codingapi/tm/api/service/ApiTxManagerService.java b/tx-manager/src/main/java/com/codingapi/tm/api/service/ApiTxManagerService.java
@@ -32,15 +32,7 @@ public interface ApiTxManagerService {
      * @param isGroup   是否合并事务
      * @return  事务状态
      */
-    boolean clearTransaction(String groupId, String taskId, int isGroup);
-
-    /**
-     * 检查事务数据
-     * @param groupId   事务组Id
-     * @param taskId    任务Id
-     * @return  事务状态
-     */
-    int getTransaction(String groupId, String taskId);
+    int cleanNotifyTransaction(String groupId, String taskId);
 
 
     /**
diff --git a/tx-manager/src/main/java/com/codingapi/tm/api/service/impl/ApiTxManagerServiceImpl.java b/tx-manager/src/main/java/com/codingapi/tm/api/service/impl/ApiTxManagerServiceImpl.java
@@ -10,7 +10,6 @@
 import com.codingapi.tm.manager.service.TxManagerService;
 import com.codingapi.tm.model.TxServer;
 import com.codingapi.tm.model.TxState;
-import com.codingapi.tm.redis.service.RedisServerService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -45,14 +44,10 @@ public TxServer getServer() {
 
 
     @Override
-    public boolean clearTransaction(String groupId, String taskId, int isGroup) {
-        return managerService.clearTransaction(groupId,taskId,isGroup);
+    public int cleanNotifyTransaction(String groupId, String taskId) {
+        return managerService.cleanNotifyTransaction(groupId,taskId);
     }
 
-    @Override
-    public int getTransaction(String groupId, String taskId) {
-        return managerService.getTransaction(groupId, taskId);
-    }
 
     @Override
     public boolean sendCompensateMsg(long currentTime, String groupId, String model, String address, String uniqueKey, String className, String methodStr, String data, int time) {
diff --git a/tx-manager/src/main/java/com/codingapi/tm/compensate/service/impl/CompensateServiceImpl.java b/tx-manager/src/main/java/com/codingapi/tm/compensate/service/impl/CompensateServiceImpl.java
@@ -57,12 +57,28 @@ 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);
+//            key = configReader.getKeyPrefixNotify() + transactionCompensateMsg.getGroupId();
+//            txGroup = redisServerService.getTxGroupByKey(key);
+            //todo  待完善
         }
         if(txGroup!=null) {
             redisServerService.deleteKey(key);
 
+            //已经全部通知的模块不做补偿处理
+            boolean hasNoNotify = false;
+            for(TxInfo txInfo:txGroup.getList()){
+                if(txInfo.getNotify()==0){
+                    hasNoNotify = true;
+                }
+            }
+
+            if(!hasNoNotify){
+                //事务已经执行完毕的
+                return true;
+            }
+
+
+
             transactionCompensateMsg.setTxGroup(txGroup);
 
             final String json = JSON.toJSONString(transactionCompensateMsg);
diff --git a/tx-manager/src/main/java/com/codingapi/tm/config/ConfigReader.java b/tx-manager/src/main/java/com/codingapi/tm/config/ConfigReader.java
@@ -43,10 +43,10 @@ public class ConfigReader {
      */
     private final String key_prefix = "tx_manager_default_";
 
-    /**
-     * 事务存在问题，临时暂存的位置，用于等待网络超时和补偿请求之前的存储位置，有最大时间
-     */
-    private final String key_prefix_notify = "tx_manager_notify_";
+//    /**
+//     * 事务存在问题，临时暂存的位置，用于等待网络超时和补偿请求之前的存储位置，有最大时间
+//     */
+//    private final String key_prefix_notify = "tx_manager_notify_";
 
     /**
      * 补偿事务永久存储数据
@@ -62,9 +62,9 @@ public String getKeyPrefix() {
         return key_prefix;
     }
 
-    public String getKeyPrefixNotify() {
-        return key_prefix_notify;
-    }
+//    public String getKeyPrefixNotify() {
+//        return key_prefix_notify;
+//    }
 
     public String getKeyPrefixCompensate() {
         return key_prefix_compensate;
diff --git a/tx-manager/src/main/java/com/codingapi/tm/manager/service/TxManagerService.java b/tx-manager/src/main/java/com/codingapi/tm/manager/service/TxManagerService.java
@@ -26,15 +26,6 @@ public interface TxManagerService {
     TxGroup addTransactionGroup(String groupId, String taskId,int isGroup, String modelName, String methodStr);
 
 
-    /**
-     *
-     * @param groupId
-     * @param taskId
-     * @return  1 存在 0不存在 -1 未结束
-     */
-    int getTransaction(String groupId, String taskId);
-
-
     boolean closeTransactionGroup(String groupId,int state);
 
 
@@ -43,7 +34,13 @@ public interface TxManagerService {
     void deleteTxGroup(TxGroup txGroup);
 
 
-    boolean clearTransaction(String groupId, String taskId, int isGroup);
+    /**
+     * 检查事务组数据
+     * @param groupId   事务组id
+     * @param taskId    任务id
+     * @return  本次请求的是否提交 1提交 0回滚
+     */
+    int cleanNotifyTransaction(String groupId, String taskId);
 
 
 
diff --git a/tx-manager/src/main/java/com/codingapi/tm/manager/service/impl/TxManagerServiceImpl.java b/tx-manager/src/main/java/com/codingapi/tm/manager/service/impl/TxManagerServiceImpl.java
diff --git a/tx-manager/src/main/java/com/codingapi/tm/netty/service/impl/ActionCKGServiceImpl.java b/tx-manager/src/main/java/com/codingapi/tm/netty/service/impl/ActionCKGServiceImpl.java
diff --git a/tx-manager/src/main/java/com/codingapi/tm/redis/service/impl/RedisServerServiceImpl.java b/tx-manager/src/main/java/com/codingapi/tm/redis/service/impl/RedisServerServiceImpl.java
