作者：京東零售 張均杰

背景

部門(mén)內(nèi)有一些億級(jí)別核心業(yè)務(wù)表增速非?？欤隽咳站?00W，但線(xiàn)上業(yè)務(wù)只依賴(lài)近一周的數(shù)據(jù)。隨著數(shù)據(jù)量的迅速增長(zhǎng)，慢SQL頻發(fā)，數(shù)據(jù)庫(kù)性能下降，系統(tǒng)穩(wěn)定性受到嚴(yán)重影響。本篇文章，將分享如何使用MyBatis攔截器低成本的提升數(shù)據(jù)庫(kù)穩(wěn)定性。

業(yè)界常見(jiàn)方案

針對(duì)冷數(shù)據(jù)多的大表，常用的策略有以2種：

刪除/歸檔舊數(shù)據(jù)。

分表。

歸檔/刪除舊數(shù)據(jù)

定期將冷數(shù)據(jù)移動(dòng)到歸檔表或者冷存儲(chǔ)中，或定期對(duì)表進(jìn)行刪除，以減少表的大小。此策略邏輯簡(jiǎn)單，只需要編寫(xiě)一個(gè)JOB定期執(zhí)行SQL刪除數(shù)據(jù)。我們開(kāi)始也是用這種方案，但此方案也有一些副作用：

1.數(shù)據(jù)刪除會(huì)影響數(shù)據(jù)庫(kù)性能，引發(fā)慢sql，多張表并行刪除，數(shù)據(jù)庫(kù)壓力會(huì)更大。

2.頻繁刪除數(shù)據(jù)，會(huì)產(chǎn)生數(shù)據(jù)庫(kù)碎片，影響數(shù)據(jù)庫(kù)性能，引發(fā)慢SQL。

綜上，此方案有一定風(fēng)險(xiǎn)，為了規(guī)避這種風(fēng)險(xiǎn)，我們決定采用另一種方案：分表。

分表

我們決定按日期對(duì)表進(jìn)行橫向拆分，實(shí)現(xiàn)讓系統(tǒng)每周生成一張周期表，表內(nèi)只存近一周的數(shù)據(jù)，規(guī)避單表過(guò)大帶來(lái)的風(fēng)險(xiǎn)。

分表方案選型

經(jīng)調(diào)研，考慮2種分表方案：Sharding-JDBC、利用Mybatis自帶的攔截器特性。

經(jīng)過(guò)對(duì)比后，決定采用Mybatis攔截器來(lái)實(shí)現(xiàn)分表，原因如下：

1.JAVA生態(tài)中很常用的分表框架是Sharding-JDBC，雖然功能強(qiáng)大，但需要一定的接入成本，并且很多功能暫時(shí)用不上。

2.系統(tǒng)本身已經(jīng)在使用Mybatis了，只需要添加一個(gè)mybaits攔截器，把SQL表名替換為新的周期表就可以了，沒(méi)有接入新框架的成本，開(kāi)發(fā)成本也不高。

分表具體實(shí)現(xiàn)代碼

分表配置對(duì)象

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Date;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class ShardingProperty {
    // 分表周期天數(shù)，配置7，就是一周一分
    private Integer days;
    // 分表開(kāi)始日期，需要用這個(gè)日期計(jì)算周期表名
    private Date beginDate;
    // 需要分表的表名
    private String tableName;
}

分表配置類(lèi)

import java.util.concurrent.ConcurrentHashMap;

public class ShardingPropertyConfig {

    public static final ConcurrentHashMap SHARDING_TABLE = new ConcurrentHashMap();

    static {
        ShardingProperty orderInfoShardingConfig = new ShardingProperty(15, DateUtils.string2Date("20231117"), "order_info");
        ShardingProperty userInfoShardingConfig = new ShardingProperty(7, DateUtils.string2Date("20231117"), "user_info");

        SHARDING_TABLE.put(orderInfoShardingConfig.getTableName(), orderInfoShardingConfig);
        SHARDING_TABLE.put(userInfoShardingConfig.getTableName(), userInfoShardingConfig);
    }
}

攔截器

import lombok.extern.slf4j.Slf4j;
import o2o.aspect.platform.function.template.service.TemplateMatchService;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.reflection.DefaultReflectorFactory;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.reflection.ReflectorFactory;
import org.apache.ibatis.reflection.factory.DefaultObjectFactory;
import org.apache.ibatis.reflection.factory.ObjectFactory;
import org.apache.ibatis.reflection.wrapper.DefaultObjectWrapperFactory;
import org.apache.ibatis.reflection.wrapper.ObjectWrapperFactory;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Properties;

@Slf4j
@Component
@Intercepts({@Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})})
public class ShardingTableInterceptor implements Interceptor {
    private static final ObjectFactory DEFAULT_OBJECT_FACTORY = new DefaultObjectFactory();
    private static final ObjectWrapperFactory DEFAULT_OBJECT_WRAPPER_FACTORY = new DefaultObjectWrapperFactory();
    private static final ReflectorFactory DEFAULT_REFLECTOR_FACTORY = new DefaultReflectorFactory();
    private static final String MAPPED_STATEMENT = "delegate.mappedStatement";
    private static final String BOUND_SQL = "delegate.boundSql";
    private static final String ORIGIN_BOUND_SQL = "delegate.boundSql.sql";
    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
    private static final String SHARDING_MAPPER = "com.jd.o2o.inviter.promote.mapper.ShardingMapper";

    private ConfigUtils configUtils = SpringContextHolder.getBean(ConfigUtils.class);

    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        boolean shardingSwitch = configUtils.getBool("sharding_switch", false);
        // 沒(méi)開(kāi)啟分表 直接返回老數(shù)據(jù)
        if (!shardingSwitch) {
            return invocation.proceed();
        }

        StatementHandler statementHandler = (StatementHandler) invocation.getTarget();
        MetaObject metaStatementHandler = MetaObject.forObject(statementHandler, DEFAULT_OBJECT_FACTORY, DEFAULT_OBJECT_WRAPPER_FACTORY, DEFAULT_REFLECTOR_FACTORY);
        MappedStatement mappedStatement = (MappedStatement) metaStatementHandler.getValue(MAPPED_STATEMENT);
        BoundSql boundSql = (BoundSql) metaStatementHandler.getValue(BOUND_SQL);
        String originSql = (String) metaStatementHandler.getValue(ORIGIN_BOUND_SQL);
        if (StringUtils.isBlank(originSql)) {
            return invocation.proceed();
        }

        // 獲取表名
        String tableName = TemplateMatchService.matchTableName(boundSql.getSql().trim());
        ShardingProperty shardingProperty = ShardingPropertyConfig.SHARDING_TABLE.get(tableName);
        if (shardingProperty == null) {
            return invocation.proceed();
        }

        // 新表
        String shardingTable = getCurrentShardingTable(shardingProperty, new Date());
        String rebuildSql = boundSql.getSql().replace(shardingProperty.getTableName(), shardingTable);
        metaStatementHandler.setValue(ORIGIN_BOUND_SQL, rebuildSql);
        if (log.isDebugEnabled()) {
            log.info("rebuildSQL -> {}", rebuildSql);
        }

        return invocation.proceed();
    }

    @Override
    public Object plugin(Object target) {
        if (target instanceof StatementHandler) {
            return Plugin.wrap(target, this);
        }
        return target;
    }

    @Override
    public void setProperties(Properties properties) {}

    public static String getCurrentShardingTable(ShardingProperty shardingProperty, Date createTime) {
        String tableName = shardingProperty.getTableName();
        Integer days = shardingProperty.getDays();
        Date beginDate = shardingProperty.getBeginDate();

        Date date;
        if (createTime == null) {
            date = new Date();
        } else {
            date = createTime;
        }
        if (date.before(beginDate)) {
            return null;
        }
        LocalDateTime targetDate = SimpleDateFormatUtils.convertDateToLocalDateTime(date);
        LocalDateTime startDate = SimpleDateFormatUtils.convertDateToLocalDateTime(beginDate);
        LocalDateTime intervalStartDate = DateIntervalChecker.getIntervalStartDate(targetDate, startDate, days);
        LocalDateTime intervalEndDate = intervalStartDate.plusDays(days - 1);
        return tableName + "_" + intervalStartDate.format(FORMATTER) + "_" + intervalEndDate.format(FORMATTER);
    }
}

臨界點(diǎn)數(shù)據(jù)不連續(xù)問(wèn)題

分表方案有1個(gè)難點(diǎn)需要解決：周期臨界點(diǎn)數(shù)據(jù)不連續(xù)。舉例：假設(shè)要對(duì)operate_log（操作日志表）大表進(jìn)行橫向分表，每周一張表，分表明細(xì)可看下面表格。

1月8號(hào)就是分表臨界點(diǎn)，8號(hào)需要切換到第二周的表，但8號(hào)0點(diǎn)剛切換的時(shí)候，表內(nèi)沒(méi)有任何數(shù)據(jù)，這時(shí)如果業(yè)務(wù)需要查近一周的操作日志是查不到的，這樣就會(huì)引發(fā)線(xiàn)上問(wèn)題。

我決定采用數(shù)據(jù)冗余的方式來(lái)解決這個(gè)痛點(diǎn)。每個(gè)周期表都冗余一份上個(gè)周期的數(shù)據(jù)，用雙倍數(shù)據(jù)量實(shí)現(xiàn)數(shù)據(jù)滑動(dòng)的效果，效果見(jiàn)下面表格。

注：表格內(nèi)第一行數(shù)據(jù)就是冗余的上個(gè)周期表的數(shù)據(jù)。

思路有了，接下來(lái)就要考慮怎么實(shí)現(xiàn)雙寫(xiě)（數(shù)據(jù)冗余到下個(gè)周期表） ,有2種方案：

1.在SQL執(zhí)行完成返回結(jié)果前添加邏輯（可以用AspectJ 或 mybatis攔截器），如果SQL內(nèi)的表名是當(dāng)前周期表，就把表名替換為下個(gè)周期表，然后再次執(zhí)行SQL。此方案對(duì)業(yè)務(wù)影響大，相當(dāng)于串行執(zhí)行了2次SQL，有性能損耗。

2.監(jiān)聽(tīng)增量binlog，京東內(nèi)部有現(xiàn)成的數(shù)據(jù)訂閱中間件DRC，讀者也可以使用cannal等開(kāi)源中間件來(lái)代替DRC，原理大同小異，此方案對(duì)業(yè)務(wù)無(wú)影響。

方案對(duì)比后，選擇了對(duì)業(yè)務(wù)性能損耗小的方案二。

監(jiān)聽(tīng)binlog數(shù)據(jù)雙寫(xiě)注意點(diǎn)

1.提前上線(xiàn)監(jiān)聽(tīng)程序，提前把老表數(shù)據(jù)同步到新的周期表。分表前只監(jiān)聽(tīng)老表binlog就可以，分表前只需要把老表數(shù)據(jù)同步到新表。

2.切換到新表的臨界點(diǎn)，為了避免丟失積壓的老表binlog，需要同時(shí)處理新表binlog和老表binlog，這樣會(huì)出現(xiàn)死循環(huán)同步的問(wèn)題，因?yàn)槔媳硇枰叫卤?，新表又需要雙寫(xiě)老表。為了打破循環(huán)，需要先把雙寫(xiě)老表消費(fèi)堵上讓消息暫時(shí)積壓，切換新表成功后，再打開(kāi)雙寫(xiě)消費(fèi)。

監(jiān)聽(tīng)binlog數(shù)據(jù)雙寫(xiě)代碼

注：下面代碼不能直接用，只提供基本思路

/**
 * 監(jiān)聽(tīng)binlog ，分表雙寫(xiě)，解決數(shù)據(jù)臨界問(wèn)題
*/
@Slf4j
@Component
public class BinLogConsumer implements MessageListener {
    
    private MessageDeserialize deserialize = new JMQMessageDeserialize();

    private static final String TABLE_PLACEHOLDER = "%TABLE%";

    @Value("${mq.doubleWriteTopic.topic}")
    private String doubleWriteTopic;

    @Autowired
    private JmqProducerService jmqProducerService;


    @Override
    public void onMessage(List messages) throws Exception {
        if (messages == null || messages.isEmpty()) {
            return;
        }
        List entryMessages = deserialize.deserialize(messages);
        for (EntryMessage entryMessage : entryMessages) {
            try {
                syncData(entryMessage);
            } catch (Exception e) {
                log.error("sharding sync data error", e);
                throw e;
            }
        }
    }

    private void syncData(EntryMessage entryMessage) throws JMQException {
        // 根據(jù)binlog內(nèi)的表名，獲取需要同步的表
        // 3種情況：
        // 1、老表：需要同步當(dāng)前周期表，和下個(gè)周期表。
        // 2、當(dāng)前周期表：需要同步下個(gè)周期表，和老表。
        // 3、下個(gè)周期表：不需要同步。
        List syncTables = getSyncTables(entryMessage.tableName, entryMessage.createTime);
        
        if (CollectionUtils.isEmpty(syncTables)) {
            log.info("table {} is not need sync", tableName);
            return;
        }

        if (entryMessage.getHeader().getEventType() == WaveEntry.EventType.INSERT) {
            String insertTableSqlTemplate = parseSqlForInsert(rowData);
            for (String syncTable : syncTables) {
                String insertSql = insertTableSqlTemplate.replaceAll(TABLE_PLACEHOLDER, syncTable);
                // 雙寫(xiě)老表發(fā)Q，為了避免出現(xiàn)同步死循環(huán)問(wèn)題
                if (ShardingPropertyConfig.SHARDING_TABLE.containsKey(syncTable)) {
                    Long primaryKey = getPrimaryKey(rowData.getAfterColumnsList());
                    sendDoubleWriteMsg(insertSql, primaryKey);
                    continue;
                }
                mysqlConnection.executeSql(insertSql);
            }
            continue;
        }
    }

數(shù)據(jù)對(duì)比

為了保證新表和老表數(shù)據(jù)一致，需要編寫(xiě)對(duì)比程序，在上線(xiàn)前進(jìn)行數(shù)據(jù)對(duì)比，保證binlog同步無(wú)問(wèn)題。

具體實(shí)現(xiàn)代碼不做展示，思路：新表查詢(xún)一定量級(jí)數(shù)據(jù)，老表查詢(xún)相同量級(jí)數(shù)據(jù)，都轉(zhuǎn)換成JSON，equals對(duì)比。

審核編輯 黃宇