compatible DatabaseSet

ctripcorp · Nov 29, 2019 · 9032ccb · 9032ccb
1 parent ac45e99
commit 9032ccb
Show file tree

Hide file tree

Showing 15 changed files with 229 additions and 63 deletions.
diff --git a/dal-client/pom.xml b/dal-client/pom.xml
@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.ctrip.platform</groupId>
     <artifactId>dal-client</artifactId>
-    <version>2.0.3</version>
+    <version>2.0.4</version>
 
     <properties>
         <java.version>1.8</java.version>

diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/ClusterDatabaseSet.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/ClusterDatabaseSet.java
@@ -19,7 +19,7 @@
 /**
  * @author c7ch23en
  */
-public class ClusterDatabaseSet implements DatabaseSet {
+public class ClusterDatabaseSet extends DatabaseSet {
 
     private String databaseSetName;
     private Cluster cluster;

diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/DatabaseSet.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/DatabaseSet.java
@@ -14,36 +14,81 @@
 import com.ctrip.platform.dal.exceptions.DalException;
 import com.ctrip.platform.dal.sharding.idgen.IIdGeneratorConfig;
 
-public interface DatabaseSet {
-
-    String getName();
-
-    String getProvider();
-
-    DatabaseCategory getDatabaseCategory();
-
-    boolean isShardingSupported();
-
-    boolean isTableShardingSupported(String tableName);
-
-    Map<String, DataBase> getDatabases();
-
-    void validate(String shard) throws SQLException;
-
-    Set<String> getAllShards();
-
-    Set<String> getAllTableShards(String tableName) throws SQLException;
-
-    DalShardingStrategy getStrategy() throws SQLException;
-
-    List<DataBase> getMasterDbs();
-
-    List<DataBase> getSlaveDbs();
-
-    List<DataBase> getMasterDbs(String shard);
-
-    List<DataBase> getSlaveDbs(String shard);
-
-    IIdGeneratorConfig getIdGenConfig();
+public abstract class DatabaseSet implements IDatabaseSet {
+
+    @Override
+    public String getName() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public String getProvider() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public DatabaseCategory getDatabaseCategory() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public boolean isShardingSupported() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public boolean isTableShardingSupported(String tableName) {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public Map<String, DataBase> getDatabases() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public void validate(String shard) throws SQLException {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public Set<String> getAllShards() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public Set<String> getAllTableShards(String tableName) throws SQLException {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public DalShardingStrategy getStrategy() throws SQLException {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public List<DataBase> getMasterDbs() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public List<DataBase> getSlaveDbs() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public List<DataBase> getMasterDbs(String shard) {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public List<DataBase> getSlaveDbs(String shard) {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
+
+    @Override
+    public IIdGeneratorConfig getIdGenConfig() {
+        throw new UnsupportedOperationException("This is an abstract DatabaseSet.");
+    }
 
 }
diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/DefaultDatabaseSet.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/DefaultDatabaseSet.java
@@ -14,7 +14,7 @@
 import com.ctrip.platform.dal.exceptions.DalException;
 import com.ctrip.platform.dal.sharding.idgen.IIdGeneratorConfig;
 
-public class DefaultDatabaseSet implements DatabaseSet {
+public class DefaultDatabaseSet extends DatabaseSet {
     private static final String CLASS = "class";
     private static final String ENTRY_SEPARATOR = ";";
     private static final String KEY_VALUE_SEPARATOR = "=";

diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/IDatabaseSet.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/configure/IDatabaseSet.java
@@ -0,0 +1,44 @@
+package com.ctrip.platform.dal.dao.configure;
+
+import com.ctrip.platform.dal.common.enums.DatabaseCategory;
+import com.ctrip.platform.dal.dao.strategy.DalShardingStrategy;
+import com.ctrip.platform.dal.sharding.idgen.IIdGeneratorConfig;
+
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public interface IDatabaseSet {
+
+    String getName();
+
+    String getProvider();
+
+    DatabaseCategory getDatabaseCategory();
+
+    boolean isShardingSupported();
+
+    boolean isTableShardingSupported(String tableName);
+
+    Map<String, DataBase> getDatabases();
+
+    void validate(String shard) throws SQLException;
+
+    Set<String> getAllShards();
+
+    Set<String> getAllTableShards(String tableName) throws SQLException;
+
+    DalShardingStrategy getStrategy() throws SQLException;
+
+    List<DataBase> getMasterDbs();
+
+    List<DataBase> getSlaveDbs();
+
+    List<DataBase> getMasterDbs(String shard);
+
+    List<DataBase> getSlaveDbs(String shard);
+
+    IIdGeneratorConfig getIdGenConfig();
+
+}
diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/datasource/DataSourceLocator.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/datasource/DataSourceLocator.java
@@ -77,6 +77,7 @@ public DataSource getDataSource(DataSourceIdentity id) {
 
     public void removeDataSource(DataSourceIdentity id) {
         DataSource ds = cache.remove(id);
+        provider.unregister(id);
         if (ds instanceof RefreshableDataSource) {
             ((RefreshableDataSource) ds).close();
         }
@@ -96,8 +97,7 @@ private DataSource createDataSource(DataSourceIdentity id) throws SQLException {
         return ds;
     }
 
-    public void setup(Cluster cluster) {
-    }
+    public void setup(Cluster cluster) {}
 
     public static Map<String, Integer> getActiveConnectionNumber() {
         Map<String, Integer> map = new HashMap<>();

diff --git a/dal-client/src/main/java/com/ctrip/platform/dal/dao/task/InsertTaskAdapter.java b/dal-client/src/main/java/com/ctrip/platform/dal/dao/task/InsertTaskAdapter.java
@@ -139,7 +139,7 @@ public void processIdentityField(DalHints hints, List<Map<String, ?>> pojos) {
 				if (identityInsertDisabled || null == pojo.get(identityFieldName)) {
 					Number id = idGenerator.nextId();
 					checkIdentityTypes(identityFieldType, id);
-					pojo.put(identityFieldName, idGenerator.nextId());
+					pojo.put(identityFieldName, id);
 				}
 			}
 		}

diff --git a/dal-cluster-client/pom.xml b/dal-cluster-client/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.ctrip.framework.dal</groupId>
     <artifactId>dal-cluster-client</artifactId>
-    <version>2.0.3</version>
+    <version>2.0.4</version>
     <!-- TODO: mvn package sources & javadoc -->
     <properties>
         <java.version>1.8</java.version>

diff --git a/...rc/main/java/com/ctrip/framework/dal/cluster/client/config/ClusterConfigXMLConstants.java b/...rc/main/java/com/ctrip/framework/dal/cluster/client/config/ClusterConfigXMLConstants.java
@@ -13,6 +13,7 @@ public interface ClusterConfigXMLConstants {
     String DATABASE = "Database";
     String SHARD_STRATEGIES = "ShardStrategies";
     String MOD_STRATEGY = "ModStrategy";
+    String USER_HINT_STRATEGY = "UserHintStrategy";
     String CUSTOM_STRATEGY = "CustomStrategy";
     String PROPERTY = "Property";
     String TABLES = "Tables";

diff --git a/...t/src/main/java/com/ctrip/framework/dal/cluster/client/config/ClusterConfigXMLParser.java b/...t/src/main/java/com/ctrip/framework/dal/cluster/client/config/ClusterConfigXMLParser.java
@@ -8,6 +8,7 @@
 import com.ctrip.framework.dal.cluster.client.sharding.idgen.ClusterIdGeneratorConfig;
 import com.ctrip.framework.dal.cluster.client.sharding.strategy.ModShardStrategy;
 import com.ctrip.framework.dal.cluster.client.sharding.strategy.ShardStrategy;
+import com.ctrip.framework.dal.cluster.client.sharding.strategy.UserHintStrategy;
 import com.ctrip.framework.dal.cluster.client.util.ServiceLoaderUtils;
 import com.ctrip.framework.dal.cluster.client.util.StringUtils;
 import org.w3c.dom.Document;
@@ -139,6 +140,8 @@ private void setAttributesForDatabase(DatabaseConfigImpl databaseConfig, Node da
     private void parseShardStrategies(ClusterConfigImpl clusterConfig, Node strategiesNode) {
         for (Node modStrategyNode : getChildNodes(strategiesNode, MOD_STRATEGY))
             parseModStrategy(clusterConfig, modStrategyNode);
+        for (Node modStrategyNode : getChildNodes(strategiesNode, USER_HINT_STRATEGY))
+            parseUserHintStrategy(clusterConfig, modStrategyNode);
         for (Node customStrategyNode : getChildNodes(strategiesNode, CUSTOM_STRATEGY))
             parseCustomStrategy(clusterConfig, customStrategyNode);
     }
@@ -147,6 +150,10 @@ private void parseModStrategy(ClusterConfigImpl clusterConfig, Node strategyNode
         parseShardStrategy(clusterConfig, strategyNode, new ModShardStrategy());
     }
 
+    private void parseUserHintStrategy(ClusterConfigImpl clusterConfig, Node strategyNode) {
+        parseShardStrategy(clusterConfig, strategyNode, new UserHintStrategy());
+    }
+
     private void parseCustomStrategy(ClusterConfigImpl clusterConfig, Node strategyNode) {
         String className = getAttribute(strategyNode, CLASS);
         try {

diff --git a/...main/java/com/ctrip/framework/dal/cluster/client/sharding/strategy/BaseShardStrategy.java b/...main/java/com/ctrip/framework/dal/cluster/client/sharding/strategy/BaseShardStrategy.java
@@ -1,6 +1,11 @@
 package com.ctrip.framework.dal.cluster.client.sharding.strategy;
 
 import com.ctrip.framework.dal.cluster.client.config.ShardStrategyElement;
+import com.ctrip.framework.dal.cluster.client.sharding.context.DbShardContext;
+import com.ctrip.framework.dal.cluster.client.sharding.context.TableShardContext;
+
+import java.util.HashSet;
+import java.util.Set;
 
 /**
  * @author c7ch23en
@@ -9,11 +14,40 @@ public abstract class BaseShardStrategy extends ShardStrategyElement implements
 
     private static final String TABLE_SHARDING = "tableSharding";
     private static final String TABLE_SHARD_SEPARATOR = "tableShardSeparator";
+    private static final String DB_SHARD_OFFSET = "dbShardOffset";
+    private static final String TABLE_SHARD_OFFSET = "tableShardOffset";
 
     private static final boolean DEFAULT_TABLE_SHARDING = false;
     private static final String DEFAULT_TABLE_SHARD_SEPARATOR = "_";
+    private static final int DEFAULT_DB_SHARD_OFFSET = 0;
+    private static final int DEFAULT_TABLE_SHARD_OFFSET = 0;
+
+    protected BaseShardStrategy() {}
+
+    protected abstract Integer calcDbShard(String tableName, DbShardContext context);
+
+    protected abstract String calcTableShard(String tableName, TableShardContext context);
+
+    @Override
+    public Integer getDbShard(String tableName, DbShardContext context) {
+        Integer shard = calcDbShard(tableName, context);
+        return shard != null ? (shard + getDbShardOffset(tableName)) : null;
+    }
+
+    @Override
+    public String getTableShard(String tableName, TableShardContext context) {
+        String shard = calcTableShard(tableName, context);
+        return shard != null ? offsetTableShard(tableName, shard) : null;
+    }
 
-    public BaseShardStrategy() {}
+    @Override
+    public Set<String> getAllTableShards(String tableName) {
+        Set<String> allShards = calcAllTableShards(tableName);
+        Set<String> offsetShards = new HashSet<>();
+        for (String shard : allShards)
+            offsetShards.add(offsetTableShard(tableName, shard));
+        return offsetShards;
+    }
 
     @Override
     public boolean tableShardingEnabled(String tableName) {
@@ -25,4 +59,23 @@ public String getTableShardSeparator(String tableName) {
         return getTableProperty(tableName, TABLE_SHARD_SEPARATOR, DEFAULT_TABLE_SHARD_SEPARATOR);
     }
 
+    protected int getDbShardOffset(String tableName) {
+        return getTableIntProperty(tableName, DB_SHARD_OFFSET, DEFAULT_DB_SHARD_OFFSET);
+    }
+
+    protected int getTableShardOffset(String tableName) {
+        return getTableIntProperty(tableName, TABLE_SHARD_OFFSET, DEFAULT_TABLE_SHARD_OFFSET);
+    }
+
+    protected Set<String> calcAllTableShards(String tableName) {
+        throw new UnsupportedOperationException(String.format("Could not calculate all table shards for table '%s'", tableName));
+    }
+
+    private String offsetTableShard(String tableName, String shard) {
+        int offset = getTableShardOffset(tableName);
+        if (offset == 0)
+            return shard;
+        return String.valueOf(Integer.parseInt(shard) + offset);
+    }
+
 }
diff --git a/...in/java/com/ctrip/framework/dal/cluster/client/sharding/strategy/ColumnShardStrategy.java b/...in/java/com/ctrip/framework/dal/cluster/client/sharding/strategy/ColumnShardStrategy.java
@@ -15,14 +15,14 @@ public abstract class ColumnShardStrategy extends BaseShardStrategy implements S
     private static final String DB_SHARD_COLUMN = "dbShardColumn";
     private static final String TABLE_SHARD_COLUMN = "tableShardColumn";
 
-    public ColumnShardStrategy() {}
+    protected ColumnShardStrategy() {}
 
-    public abstract Integer calcDbShard(String tableName, Object shardValue);
+    protected abstract Integer calcDbShard(String tableName, Object shardValue);
 
-    public abstract String calcTableShard(String tableName, Object shardValue);
+    protected abstract String calcTableShard(String tableName, Object shardValue);
 
     @Override
-    public Integer getDbShard(String tableName, DbShardContext context) {
+    protected Integer calcDbShard(String tableName, DbShardContext context) {
         Integer shard = context.getShardId();
         if (shard != null)
             return shard;
@@ -50,7 +50,7 @@ public Integer getDbShard(String tableName, DbShardContext context) {
     }
 
     @Override
-    public String getTableShard(String tableName, TableShardContext context) {
+    protected String calcTableShard(String tableName, TableShardContext context) {
         String shard = context.getShardId();
         if (shard != null)
             return shard;
@@ -77,7 +77,7 @@ public String getTableShard(String tableName, TableShardContext context) {
         return shard;
     }
 
-    private Object getDbShardValue(String tableName, ShardData shardData) {
+    protected Object getDbShardValue(String tableName, ShardData shardData) {
         if (shardData == null)
             return null;
         String dbShardColumn = getTableProperty(tableName, DB_SHARD_COLUMN);
@@ -86,7 +86,7 @@ private Object getDbShardValue(String tableName, ShardData shardData) {
         return shardData.getValue(dbShardColumn);
     }
 
-    private Object getTableShardValue(String tableName, ShardData shardData) {
+    protected Object getTableShardValue(String tableName, ShardData shardData) {
         if (shardData == null)
             return null;
         String tableShardColumn = getTableProperty(tableName, TABLE_SHARD_COLUMN);