Mybatis动态分表的流程抽象

今天药忘吃喽~ 2022-05-27 12:10 187阅读 0赞

动态分表，意思就是按照一定的拆分标准，最典型的就是时间，每隔一段时间新生成一张表，或者在插入时计算此记录是否在现有表的插入区间之内，如果不在那么立即创建新表，然后将记录插入新表中。查询时根据查询条件计算记录分布的子表集合，使用UNION多表查询之后统一分组，排序，分页；更新和删除也可自动计算符合条件的子表集合，统一操作。

以上操作还是有一些共性可抽取：

1.  动态的创建表。比如通过定时任务定期创建；或者在插入时实时计算，若新记录不属于现有区间，那么实时生成新的表，然后插入此记录。
2.  都需要实时计算操作的子表，可能是一张，也可能是多张。

我对上述流程做了抽象，基于Mybatis的形式，在实际操作前，先说明下Mybatis的动态分表实现。

**Mybatis的Insert，Update，Delete，Select这些xml元素节点，statementType默认值是PREPARED，也就是预编译的。这种类型的XML节点解析生成的Java对象是PreparedStatement类型，其在执行SQL前会先将此SQL字符串通过Connection传入数据库进行预编译，预先指定相关的表名，字段名称，这样在实际执行时，能够防止SQL注入**。

但是动态分表的情况下，因为不确定实际操作的表名，所以预编译不符合场景了，所以要指定statementType类型为STATEMENT：

这种情况下执行时没有预编译的前置步骤，所以不能防止SQL注入，但在某些场景下仍能发挥作用。因为没有预编译的步骤，所以其在实际执行时是将SQL字符串直接传入数据库执行。

我们在使用Mybatis时，大部分情况下使用 “\#\{ \}” 的形式来定义参数占位符，Mybatis在解析时会将“\#\{ \}”内的内容替换成“？”, 如果有预编译过程，那么先对SQL字符串传入数据库编译，然后将参数按顺序传入数据库执行。如果没有预编译过程，那么直接以“？”占位符的形式传入数据库执行，这显然是不符合我们当前的场景的。所以动态分表相关的SQL中不能含有“\#\{ \}”。

Mybatis还提供了 “￥\{ \}” (csdn博客问题) 形式的参数占位符定义形式。在这种形式下，如果参数是基本数据类型，那么使用”$ \{value\}”，Mybatis在解析时能将参数替换”￥\{ \}”，拼接到SQL字符串中；如果是复杂类型的参数， 使用形式和“\#\{ \}”一致。

Mybatis在解析“$ \{ \}”时，将“\{ \}”里的内容提取出来，如果发现时“value”，且参数是基本你数据类型，那么获取其值，拼接到SQL字符串中。如果不是“value”，那么根据“\{ \}”里的内容从参数中提取此属性对应的值，追加到SQL字符串中。也就是说“\{ \}”里面不能包含其他元素，比如javaType，jdbcType，typeHandler等等，若有则会影响属性值的提取。**也就说通过拼接字符串生成SQL的过程中，“\{ \}”里面只能包含属性名称或者“value”**，当然在查询到结果后的解析过程中，typeHandler等元素还是能正常发挥作用的。

将属性值从参数对象中提取出来，然后追加到SQL字符串中，最后生成的SQL字符串格式可能如下：

select * from user where name = uuu and age > 20

Mybatis就会把这个字符串拿到数据库执行，然后就出错了。因为在拼接时，字符串片段和数值片段之间没有明显的区别，没有用引号标识查询参数，执行出错。**所以在设置字符串类型的属性值时，需要在首尾两端加上引号**。

**还有一点，如果参数类型是日期，肯定不能用Date对象来拼接字符串，你需要将其转换为字符串，使用DateFormat来格式化，然后在首尾两端加上引号，这样的形式在数据库内是能被自动识别为日期而加以使用的**。

**总上所述，String属性的首尾追加引号，Date属性的格式化后追加引号，这一点也能作为共性抽取出来。**

下面贴上我自己的抽象代码：

/** * 分表业务接口 * * @author wb-jjb318191 * @create 2018-04-17 13:00 */
    public interface TableSplittingService { 
    
        /** * 指定拆分时间间隔,单位:月 * * @return */
        int getSplitIntervalInMonth();
    
        /** * 指定时间格式化形式 * * @return */
        String getDateFormatPattern();
    
        /** * 以时间为分割依据时,查询分表内时间的最大值和最小值 * * @param order 当前分表的序号 * @return */
        List<Date> selectTimeEndpoint(int order);
    
        /** * 根据当前分表的序号生成分表名称,从0开始 * * @param order * @return */
        String generateTableNameByOrder(int order);
    
        /** * 创建分表 * * @param order 当前分表的序号,从0开始 */
        void createSplitTable(int order);
    
        /** * 提取当前插入记录的插入依据,以此作为插入哪张表 * * @param obj * @return */
        Date extractInsertBasis(Object obj);
    
        /** * 获取查询对象的时间区间 * 第一个元素存放起始时间 * 第二个元素存放终止时间 * * @param obj * @return */
        List<Date> getQueryTimeInterval(Object obj);
    
    }

对外暴露一个接口，当要对指定表做拆分时需要实现此接口，此处默认按时间拆分。

@Service
    public class TypeAttrInstanceHistorySplitServiceImpl implements TableSplittingService { 
    
        @Autowired
        private TypeAttrInstanceHistoryDAO historyDAO;
    
        @Autowired
        private DiamondService diamondService;
    
        /** * diamond的分表时间间隔配置项名称 */
        private static final String ATTR_HISTORY_SPLIT_INTERVAL_IN_MONTH = "attrHistorySplitIntervalInMonth";
    
        private static final String DATE_FORMAT_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS";
    
        private static final String RAW_TABLE_NAME = "TYPE_ATTR_INSTANCE_HISTORY";
    
        @Override
        public int getSplitIntervalInMonth() {
            return diamondService.getInteger(ATTR_HISTORY_SPLIT_INTERVAL_IN_MONTH);
        }
    
        @Override
        public String getDateFormatPattern() {
            return DATE_FORMAT_PATTERN;
        }
    
        @Override
        public void createSplitTable(int order) {
            String newTableName = generateTableNameByOrder(order);
            historyDAO.createSplitTable(newTableName);
            historyDAO.addPrimaryKey(newTableName);
            historyDAO.createIndex(newTableName, "attr_history_space_id_index_" + order, "space_id");
            historyDAO.setPrimaryAutoIncrement(newTableName);
        }
    
        @Override
        public List<Date> selectTimeEndpoint(int order) {
            return historyDAO.selectTimeEndpointIfExist(generateTableNameByOrder(order));
        }
    
        @Override
        public String generateTableNameByOrder(int order) {
            return order == 0 ? RAW_TABLE_NAME : RAW_TABLE_NAME + "_" + order;
        }
    
        @Override
        public Date extractInsertBasis(Object obj) {
            return ((TypeAttrInstanceHistory)obj).getGmtModified();
        }
    
        @Override
        public List<Date> getQueryTimeInterval(Object obj) {
            TypeAttrInstanceHistoryQuery query = (TypeAttrInstanceHistoryQuery)obj;
            return Arrays.asList(query.getStartDate(), query.getEndDate());
        }
    
    }

import java.lang.reflect.Field;
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.Calendar;
    import java.util.Collection;
    import java.util.Date;
    import java.util.HashMap;
     import java.util.HashSet;
    import java.util.List;
    import java.util.Map;
    import java.util.Map.Entry;
    import java.util.Set;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentMap;
    
    import com.google.gson.Gson;
    import com.google.gson.GsonBuilder;
    import org.json.JSONObject;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.core.ResolvableType;
    import org.springframework.stereotype.Component;
    import org.springframework.util.Assert;
    import org.springframework.util.ReflectionUtils;
    
    /** * 分表业务实现抽象 * * @author wb-jjb318191 * @create 2018-04-16 9:48 */
    @Component
    public class TableSplittingManager implements InitializingBean { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(TableSplittingManager.class);
    
        private Gson gson;
    
        private int splitInterval;
    
        private TableSplittingService splittingService;
    
        private volatile int latestOrder = 0;
    
        private static final ConcurrentMap<Integer, SplitTable> ORDER_TO_TABLE_MAPPINGS = new ConcurrentHashMap<Integer, SplitTable>();
        /** * 插入数据的属性名称 >> 属性类型的映射,目的是为了方便对字符串字段值前后加引号 */
        private static final ConcurrentMap<Class, Map<String, Class<?>>> FIELD_TO_TYPE_MAPPINGS = new ConcurrentHashMap<Class, Map<String, Class<?>>>();
    
        /** * 查询所有分表,记录表名和数据起始,结束时间的映射 */
        @Override
        public void afterPropertiesSet() {
            Assert.notNull(splittingService, "[splittingService] 属性不能为空");
    
            splitInterval = splittingService.getSplitIntervalInMonth();
            gson = new GsonBuilder().setDateFormat(splittingService.getDateFormatPattern()).create();
    
            do {
                try {
                    List<Date> endpoint = splittingService.selectTimeEndpoint(latestOrder);
                    String tableName = splittingService.generateTableNameByOrder(latestOrder);
                    ORDER_TO_TABLE_MAPPINGS.put(latestOrder++, buildSplitTable(tableName, endpoint));
                } catch (Exception e) {
                    latestOrder--;
                    break;
                }
            } while (true);
        }
    
        /** * 构建Insert参数集合,table将存放表名 * 其他的以属性名称为key,属性值为value * 非数字类型的value两端加上单引号 * * @param record * @return */
        public Map<String, Object> buildInsertParamMap(Object record) {
            Map<String, Object> paramMap = buildParamMap(record);
            paramMap.put("table", getInsertTable(record));
            return paramMap;
        }
    
        /** * 创建查询时的参数Map集合 * 查询可能涉及多张表,用tables属性存储表名集合 * * @param query * @return */
        public Map<String, Object> buildQueryParamMap(Object query) {
            Map<String, Object> paramMap = buildParamMap(query);
            paramMap.put("tables", getSelectedTables(query));
            return paramMap;
        }
    
        /** * 获取当前待插入数据的表名 * * @param record * @return */
        private String getInsertTable(Object record) {
            int order = latestOrder;
            Date gmtModified = splittingService.extractInsertBasis(record);
            SplitTable latest = ORDER_TO_TABLE_MAPPINGS.get(order);
            if (gmtModified.before(latest.getEndTime())) {
                return latest.getTableName();
            }
            // 创建新的分表
            if (order == latestOrder) {
                synchronized (this) {
                    if (order == latestOrder) {
                        createNewTable(latest.getEndTime());
                    }
                }
            }
            return getInsertTable(record);
        }
    
        /** * 获取当前查询时间对应的分表 * 若未指定查询时间区间,则默认查询所有分表 * * @param query * @return */
        private List<String> getSelectedTables(Object query) {
            List<Date> timeEndpoint = splittingService.getQueryTimeInterval(query);
            Date startDate = timeEndpoint.get(0);
            Date endDate = timeEndpoint.get(1);
            List<String> tables = new ArrayList<String>(ORDER_TO_TABLE_MAPPINGS.size());
            for (Entry<Integer, SplitTable> entry : ORDER_TO_TABLE_MAPPINGS.entrySet()) {
                SplitTable table = entry.getValue();
                boolean tooLater = startDate != null && startDate.after(table.getEndTime());
                boolean tooEarly = endDate != null && endDate.before(table.getStartTime());
                if (!tooEarly && !tooLater) {
                    tables.add(splittingService.generateTableNameByOrder(entry.getKey()));
                }
            }
            return tables;
        }
    
        /** * 创建新的分表 * * @param start 起始时间 */
        private void createNewTable(Date start) {
    
            Thread thread = new Thread() {
    
                // 当Service层环绕事务时,在Service方法内创建表,事务未提交,新表还未实际生成。
                // 所以以异步形式跳出当前事务,非事务状态下创建表及设置索引等操作。
                @Override
                public void run() {
                    splittingService.createSplitTable(++latestOrder);
                }
            };
    
            thread.start();
            try {
                thread.join();
                SplitTable splitTable = new SplitTable(splittingService.generateTableNameByOrder(latestOrder), start, computeEndTime(start));
                ORDER_TO_TABLE_MAPPINGS.put(latestOrder, splitTable);
            } catch (InterruptedException e) {
    
            }
        }
    
        /** * 生成分表内置对象 * * @param tableName * @param endpoint * @return */
        private SplitTable buildSplitTable(String tableName, List<Date> endpoint) {
            if (endpoint.isEmpty()) {
                return new SplitTable(tableName, new Date(), computeEndTime(new Date()));
            } else {
                Date expectEnd = computeEndTime(endpoint.get(0));
                Date actualEnd = endpoint.get(1);
                // 如果分表内最近更新时间在时间区间之内,调整分表的结束时间
                if (actualEnd.before(expectEnd)) {
                    actualEnd = expectEnd;
                }
                return new SplitTable(tableName, endpoint.get(0), actualEnd);
            }
        }
    
        /** * 根据起始时间计算结束时间 * * @param startTime * @return */
        private Date computeEndTime(Date startTime) {
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(startTime);
            calendar.add(Calendar.MONTH, splitInterval);
            return calendar.getTime();
        }
    
        /** * 构建待插入数据的参数列表,主要是需要对字符串类型,时间类型的字段前后加上引号 * * @param object * @return */
        private Map<String, Object> buildParamMap(Object object) {
            Map<String, Object> paramMap = new JSONObject(gson.toJson(object)).toMap();
            Class<?> clazz = object.getClass();
            Map<String, Class<?>> name2TypeTable = FIELD_TO_TYPE_MAPPINGS.get(clazz);
            if (name2TypeTable == null) {
                name2TypeTable = buildFieldTypeMapping(clazz);
                FIELD_TO_TYPE_MAPPINGS.putIfAbsent(clazz, name2TypeTable);
            }
            for (Entry<String, Object> entry : paramMap.entrySet()) {
                String fieldName = entry.getKey();
                Class<?> fieldType = name2TypeTable.get(fieldName);
                // 对String和Date格式化后的字符串,在首尾两端添加引号
                if (String.class == fieldType || Date.class == fieldType) {
                    entry.setValue("\'" + entry.getValue() + "\'");
                }
                //对Boolean类型转换为数值
                if (Boolean.class == fieldType) {
                    entry.setValue((Boolean)entry.getValue() ? 1 : 0);
                }
                //如果当前属性是集合或者数组,取回属性值,因为此属性可能需要在XML内做forEach操作
                if (Collection.class.isAssignableFrom(fieldType) || fieldType.isArray()) {
                    try {
                        Field field = ReflectionUtils.findField(clazz, fieldName);
                        field.setAccessible(true);
                        Object value = field.get(object);
                        // 只对字符串的集合，数组做处理,忽略日期
                        if (String.class == ResolvableType.forField(field).resolveGeneric(0)) {
                            value = processStringValues(value);
                        }
                        entry.setValue(value);
                    } catch (Exception e) {
    
                    }
                }
            }
            return paramMap;
        }
    
        /** * 将String集合，数组属性内的值前后追加引号 * * @param values * @return */
        private Object processStringValues(Object values) {
            if (values.getClass().isArray()) {
                String[] raw = (String[])values;
                String[] strings = new String[raw.length];
                for (int i = 0; i < raw.length; i++) {
                    strings[i] = "\'" + raw[i] + "\'";
                }
                return strings;
            }
            List<String> strings = new ArrayList<>();
            for (String value : (Collection<String>)values) {
                strings.add("\'" + value + "\'");
            }
            return strings;
        }
    
        /** * 构建指定类型的属性名称和属性类型的映射 * * @param clazz * @return */
        private Map<String, Class<?>> buildFieldTypeMapping(Class<?> clazz) {
            Set<Field> fields = new HashSet<Field>();
            do {
                fields.addAll(new HashSet<Field>(Arrays.asList(clazz.getDeclaredFields())));
                clazz = clazz.getSuperclass();
            } while (clazz != Object.class);
            Map<String, Class<?>> name2TypeTable = new HashMap<String, Class<?>>();
            for (Field field : fields) {
                name2TypeTable.put(field.getName(), field.getType());
            }
            return name2TypeTable;
        }
    
        public void setSplittingService(TableSplittingService splittingService) {
            this.splittingService = splittingService;
        }
    
        /** * 分表的基础数据信息 */
        static class SplitTable {
    
            private String tableName;
            private Date startTime;
            private Date endTime;
    
            public SplitTable(String tableName, Date startTime, Date endTime) {
                this.tableName = tableName;
                this.startTime = startTime;
                this.endTime = endTime;
            }
    
            public String getTableName() {
                return tableName;
            }
    
            public void setTableName(String tableName) {
                this.tableName = tableName;
            }
    
            public Date getStartTime() {
                return startTime;
            }
    
            public void setStartTime(Date startTime) {
                this.startTime = startTime;
            }
    
            public Date getEndTime() {
                return endTime;
            }
    
            public void setEndTime(Date endTime) {
                this.endTime = endTime;
            }
        }
    
    }

使用时需要实现TableSplittingService接口，实现其一系列的方法，然后注入到TableSplittingManager中。

TableSplittingManager在实例化时，会查询所有子表

/** * 以时间为分割依据时,查询分表内时间的最大值和最小值 * * @param order 当前分表的序号 * @return */
        List<Date> selectTimeEndpoint(int order);
    
        /** * 根据当前分表的序号生成分表名称,从0开始 * * @param order * @return */
        String generateTableNameByOrder(int order);

通过上述方法，从序号0开始依次查询子表的时间区间，实现类需要根据序号order，生成子表名，然后查询此子表的时间区间，比如下面的SQL：

<select id="selectTimeEndpointIfExist" parameterType="java.lang.String" resultType="java.util.Date" statementType="STATEMENT">
            SELECT MIN(GMT_MODIFIED) FROM ${value}
            UNION ALL
            SELECT MAX(GMT_MODIFIED) FROM ${value}
        </select>

TableSplittingManager按序号获取子表的时间区间，存储表名和时间区间的数据。直到发生异常，也就是子表不存在。

/** * 指定时间格式化形式 * * @return */
        String getDateFormatPattern();
    
        /** * 提取当前插入记录的插入依据,以此作为插入哪张表 * * @param obj * @return */
        Date extractInsertBasis(Object obj);
    
        /** * 指定拆分时间间隔,单位:月 * * @return */
        int getSplitIntervalInMonth();

通过实现此方法来指定日期的格式化格式，比如：yyyy-MM-dd HH:mm:ss

通过extractInsertBasis方法，将待插入数据的时间依据提取出来，插入时会根据此时间来决定时插入现有表还是建新表后插入

需要通过getSplitIntervalInMonth方法指定拆分区间，这样新建的表就能够指定结束区间

/** * 构建Insert参数集合,table将存放表名 * 其他的以属性名称为key,属性值为value * 非数字类型的value两端加上单引号 * * @param record * @return */
        public Map<String, Object> buildInsertParamMap(Object record) {
            Map<String, Object> paramMap = buildParamMap(record);
            paramMap.put("table", getInsertTable(record));
            return paramMap;
        }

TableSplittingManager对外暴露格式化插入对象的方法，将插入对象转换成一个Map< String,Object>集合，日期格式化为字符串，Boolean转换为0,1，字符串首尾加引号，通过根据当前记录的时间来决定插入的表名。

若表不存在，那么你需要实现创建表的方法，在其内部调用一些列的SQL方法，比如创建表，增加索引，设置主键自增等等

/** * 创建分表 * * @param order 当前分表的序号,从0开始 */
        void createSplitTable(int order);
    
    
        <update id="createSplitTable" parameterType="java.lang.String" statementType="STATEMENT">
            CREATE TABLE ${value} (
                ID INT8  NOT NULL,
                GMT_MODIFIED TIMESTAMP(6),
                ......
            )
            WITH (OIDS=FALSE)
        </update>
    
        <update id="addPrimaryKey" parameterType="java.lang.String" statementType="STATEMENT">
            ALTER TABLE ${value} ADD PRIMARY KEY ("id")
        </update>
    
        <update id="createIndex" statementType="STATEMENT">
            CREATE INDEX ${index} ON ${tableName} USING BTREE (${column})
        </update>
    
        <update id="setPrimaryAutoIncrement" parameterType="java.lang.String" statementType="STATEMENT">
            ALTER TABLE ${value} ALTER COLUMN ID SET DEFAULT NEXTVAL('type_attr_instance_history_id_seq')
        </update>

然后将表名存入Map的“table”属性中，之后就可以执行插入操作，比如：

<insert id="insertSelective" parameterType="java.util.HashMap" statementType="STATEMENT">
            insert into ${table}
            <trim prefix="(" suffix=")" suffixOverrides=",">
                <if test="gmtModified != null">
                    gmt_modified,
                </if>
                <if test="modifier != null">
                    modifier,
                </if>
                ......
            </trim>
            <trim prefix="values (" suffix=")" suffixOverrides=",">
                <if test="gmtModified != null">
                    ${gmtModified},
                </if>
                <if test="modifier != null">
                    ${modifier},
                </if>
                ......
            </trim>
        </insert>

TableSplittingManager对外暴露格式化查询对象的方法，将查询对象格式化为Map< String，Object>.

首先根据提取查询参数的查询时间区间：

/** * 获取查询对象的时间区间 * 第一个元素存放起始时间 * 第二个元素存放终止时间 * * @param obj * @return */
        List<Date> getQueryTimeInterval(Object obj)

然后根据区间来计算查询的记录分布在哪些子表中，将子表的名称以tables的key存入Map中：

/** * 创建查询时的参数Map集合 * 查询可能涉及多张表,用tables属性存储表名集合 * * @param query * @return */
        public Map<String, Object> buildQueryParamMap(Object query) {
            Map<String, Object> paramMap = buildParamMap(query);
            paramMap.put("tables", getSelectedTables(query));
            return paramMap;
        }

当然Date，String，Boolean，还有List< String>类似的属性都会做相应的转换。

之后就能根据查询的参数集合做查询操作，比如：

<select id="selectFromTables" parameterType="java.util.HashMap" resultMap="VOResultMap">
            SELECT TEMP......
            FROM
            <if test="tables != null and tables.size() > 0">
                <foreach collection="tables" open="(" item="table" separator="UNION" close=") AS TEMP">
                    SELECT ...... FROM ${table} JOIN ... ON ... WHERE ...
                </foreach>
            </if>
            ORDER BY TEMP...
            LIMIT ${limit} OFFSET ${startRow}
        </select>

查询的记录可能分布在多张表，所以需要UNION的形式先获取所有记录，生成临时表，然后做统一的分组，排序和分页等。

修改和删除的参数对象转换没有写，感兴趣的可以自行拓展，流程应该差不多的。

以上就是动态分表的流程抽象，做个总结：

1.  **每次启动时会实时加载所有现存的子表，以备实时计算**
2.  **实时计算操作涉及的子表，当子表还不存在时，实时创建**
3.  **对插入对象做格式转换，都转换为Map< String，Object>形式，同时对Date，String，Boolean形式的属性做相应的处理**