分库分表修改

src/main/java/com/dsk/flink/dsc/common/function/AsyncMysqlDataTransferFunction.java

 package com.dsk.flink.dsc.common.function;
 
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.StrUtil;
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.dsk.flink.dsc.utils.EnvProperties;
+import org.apache.commons.lang3.ObjectUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
@@ -66,23 +69,45 @@ public class AsyncMysqlDataTransferFunction extends RichAsyncFunction<JSONObject
                     pkNames.forEach(name -> pkNameSet.add(String.valueOf(name)));
                 }
 
-                String shardedTable = dbInfoMap.getSharded_table();
-                List<Map<String,Object>> list = JSONArray.parseObject(shardedTable, List.class);
-
-                List<Map<String, Object>> table1 = list.stream().filter(l -> {
-                    return l.get("table").toString().equals(table);
-                }).collect(Collectors.toList());
-
+                String shardingRule = dataObj.getString("shardingRule");
                 String concat = "";
-                //分表数
-                String tbNum = table1.get(0).get("tb_num").toString();
-                //分表字段
-                Integer majorKey = dataObj.getInteger(table1.get(0).get("major_key").toString());
-                if (!tbNum.equals("1")){
-                    int i1 = majorKey.intValue() % Integer.parseInt(tbNum);
-                    concat = table.concat("_").concat(String.valueOf(i1));
+                if (StrUtil.isNotBlank(shardingRule)){
+                    Map<String,Object> map = JSON.parseObject(shardingRule, Map.class);
+                    String strategy = map.get("strategy").toString();
+
+                    Map<String,Object> strategyMap = JSON.parseObject(strategy, Map.class);
+                    String str = MapUtil.getStr(strategyMap, "sharding-column");
+                    int i = MapUtil.getInt(strategyMap, "sharding-count").intValue();
+
+                    if (i > 1){
+                        int tableNum = dataObj.getInteger(str).intValue() % i;
+                        concat = table.concat("_").concat(String.valueOf(tableNum));
+                    }else {
+                        concat = table;
+                    }
+
                 }else {
-                    concat = table;
+                    String shardedTable = dbInfoMap.getSharded_table();
+                    List<Map<String,Object>> list = JSONArray.parseObject(shardedTable, List.class);
+
+                    List<Map<String, Object>> table1 = list.stream().filter(l -> {
+                        return l.get("table").toString().equals(table);
+                    }).collect(Collectors.toList());
+
+                    if (ObjectUtils.isEmpty(table1)){
+                        concat = table;
+                    }else {
+                        //分表数
+                        String tbNum = table1.get(0).get("tb_num").toString();
+                        //分表字段
+                        Integer majorKey = dataObj.getInteger(table1.get(0).get("major_key").toString());
+                        if (!tbNum.equals("1")){
+                            int i1 = majorKey.intValue() % Integer.parseInt(tbNum);
+                            concat = table.concat("_").concat(String.valueOf(i1));
+                        }else {
+                            concat = table;
+                        }
+                    }
                 }
 
                 String excueteSql = "";