如何做到数据变更的自动化?

为什么会有数据变更

在以平台为级别的软件集合中,为了保证软件质量的可控,错误的可追溯,不可避免的会通过一些流程来约束、框定公司的各种条条框框,比如发布需要有流程才能发布,业界有JIRAITSM这种优秀的集成式软件来制定流程。

而其中数据变更也需要一个流程来控制,当线上需要发生数据变更的时候,团队需要先编写数据库脚本,编写完后提一个流程到DBA这边,DBA执行以后会告诉提变更的人具体变更的行数有多少,确认后DBA再做COMMIT,执行完后确认变更结果,至此一整个数据变更的业务流就完成了。

而这一切的一切都是需要人来参与的,人就代表着会出错,会有沟通成本,这显然不符合如今敏捷开发的思想,更何况你的生产网往往与办公网段处于物理隔离的时候,这种变更流程所带来的时间和空间成本变得比较大,很多很急的数据变更不能及时的相应,最后甚至会有平台内部的一些子团队会分配一个人力去专门提流程、发布。如果能把这个现有的流程做提取,自动化之后,团队的成本就能进一步的降低,解放出来的DBA也能够做更有意义的事情。

有了问题,开发人员和架构师就会对应参与进来,企图通过软件的手段来解决这个问题,在与DBA充分沟通、整理了大致的业务流之后,发现其实核心的业务执行流程很简单,下面就简单的列出如下。

数据变更的核心业务流

初步方案

本着又不是不能用的心境,先写个最简单的咯,其实就是简单的JDBC的执行

// 录入端传入decSqlListStr
// delete from tableName1; delete from tableName2;
// 简易实现
String decSqlListStr = decSqlListStr.trim();
String[] sqlArr = decSqlListStr.split("\\n*\\s*;\\n*\\s*");
int[] count = jdbcTemplate.batchUpdate(); // 返回批量内容

这个例子直接返回了允许结果集,用户可以直接查看对应的影响行数。

改进方案

上述的实现是最初的简单想法,很显然和上方的期望业务流比只做到了一部分,并且不提供回滚策略,但至少做到了数据变更自动化,可以不需要人参与自动执行了,但现今的这个操作是一个很危险的操作,对于提数据变更的人可能会出现不敢做变更的情况,典型的执行出去的SQL就像泼出去的水,再也回不来了!

其实只要对他的数据变更做一定的控制即可?这就诞生了第二个思路:录入数据影响范围,如果最终执行的数据变更不在提交的范围内部的话,程序会抛出异常,Spring框架的特性导致了抛出异常之后可以自动回滚。增量的代码如下所示。

// 在执行executeBatch下执行代码块
Long allCount = 0L;
for (int i = 0; i < resCount.length; i++) {
  allCount += resCount[i];
}
if (allCount < min && allCount > max) {  // min、max为传入的录入参数
  throw new EasiSqlException("您的语句总影响行数超过配置行数,无法操作,系统自动回滚");
}

这么做是一种折中的解决方案,来做到应用层的相对可控、可回滚。

但这么做也有一个问题,既然如此,执行数据变更的用户都尽量大了填,即使要做到数据变更范围不瞎填,还要在变更前count一下,那就失去了做自动化变更的意义了,那有没有一种完美的方案,来充分的做到上方画的业务流图呐。像DBA最常用的ToadNavicat都是怎么做分段式提交的呐?因为Java都是同步的,无法做异步获取提交。那其实这个业务流最核心的问题就是如何做到分段式的执行和提交。

分段式提交

数据库客户端在执行成功后会返回影响行数,然后再等待DBAcommit或者rollback,这种分段式提交可以让用户做二段式确认,可以让执行有回旋的余地。也是数据变更最常用的一种方案。

最终的方案(实现)

分段就代表了异步,说到Java的异步,就想到了异步的线程执行数据变更,当执行指令发出去后,建立一个数据库的提交线程,这个线程包含数据库连接、执行的语句、影响行数、线程ID号、开始时间、是否终止以及提交状态

数据库连接connection

数据库连接的缓存可以做到让数据库进行分段式的数据提交

影响行数excuteCount

执行完发送notify事件的时候可以发送影响的行数,便于操作人员判断

线程ID号threadId

标识了内容,保证notify事件的唯一性

开始时间statTimestemp

记录开始事件,来释放因为业务原因导致没触发提交、回滚的资源

是否终止end

由于机器的资源有限,因此需要定期清理执行完的线程,终止标识可以让清理程序抓去到后释放线程池的资源。

提交状态commitLevel

此为异步的核心,决定了一个数据变更事件是否提交的标识,当数据变更在数据库层执行完成后,线程定期检查状态标识,不同的标识对应不同的策略,分为三种状态:

  1. hold的线程等待状态,此时等待着指令触发提交或者回滚,此时会一直等待指令的触发
  2. commit的提交状态,此时触发数据提交
  3. rollback的回滚状态,此时触发数据执行
/***
 * 数据库层面的commit等待线程
 */
public class DBCommitThread extends Thread {

    private final Log LOG = LogFactory.getLog(ClearDBCommintTaskConfig.class);

    private final static int MAX_RETRY_TIME = 5;

    // 半小时不释放,自动回滚(毫秒)
    private final static Integer EXCUTE_MAX_MINUTE = 1800000;

    // JDBC执行
    private Connection connection;

    // 执行的SQL语句
    private String[] sqlList;

    private String threadId;

    private Long statTimestemp;

    private int[] excuteCount;

    private int retryTime = 0;

    public boolean end = false;

    // 0: hold, 1: commit, 2: rollback
    private String commitLevel = "0";

    public DBCommitThread (Connection connection, String[] sqlList, String threadId) {
        this.connection = connection;
        this.sqlList = sqlList;
        this.threadId = threadId;
    }

    @Override
    public void run () {
        Statement stmt = null;
        try {
            connection.setAutoCommit(false);
            statTimestemp = System.currentTimeMillis();
            stmt = connection.createStatement();
            for (int i = 0; i < sqlList.length; i++) {
                stmt.addBatch(sqlList[i]);
            }
            if (stmt != null) {
                // 执行
                int[] count = stmt.executeBatch();
                excuteCount = count;
                notifyToServer("1", null);

                while ("0".equals(this.commitLevel)) {
                    sleep(2000);
                    Long nowExcuteTime = System.currentTimeMillis() - statTimestemp;
                    if (DBCommitThread.EXCUTE_MAX_MINUTE < nowExcuteTime) {
                        // holding too long, rollback
                        this.commitLevel = "2";
                    }
                }
                // commit or rollback
                if ("1".equals(this.commitLevel)) {
                    LOG.info("------BATCH DB COMMIT------");
                    connection.commit();
                } else {
                    LOG.info("------BATCH DB ROLLBACK------");
                    connection.rollback();
                }
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            notifyToServer("0", e.getMessage());
            rollback();
        } finally {
            close(stmt, connection);
        }

        // End Thread
        if (!isInterrupted()) {
            isInterrupted();
            this.end = true;
        }
    }

    /***
     * outer call
     * @param commitLevel commit level
     */
    public void setCommitLevel (String commitLevel) {
        this.commitLevel = commitLevel;
    }

    public String getExcuteCount () {
        return JsonUtils.beanToJson(excuteCount);
    }

    private void rollback () {
        try {
            connection.rollback();
        } catch (SQLException se) {
            LOG.error(se.getMessage(), se);
        }
    }

    private void close (Statement stmt, Connection connection) {
        try {
            if (null != stmt) {
                stmt.close();
            }
            if (null != connection && !connection.isClosed()) {
                connection.setAutoCommit(true);
                connection.close();
            }
        } catch (SQLException se) {
            LOG.error(se.getMessage(), se);
        }
    }

    private void notifyToServer (String successStatus, String errMsg) {
      // 通知主程序更新数据变更状态,为第一张业务流程图里的第二级,此处实现不做展示
    }

}

线程池方法实现如下,清理Task为直接调用clearEndThread不做展示

public class ThreadFactory {

    public static Map<String, DBCommitThread> dbCommitThreadMap;

    static {
        ThreadFactory.dbCommitThreadMap = new ConcurrentHashMap<>();
    }

    // 创建线程,数据变更开始建立
    public static String createDBCommitThread (Connection connection, String[] sqlList) {
        // 超出内容
        if (dbCommitThreadMap.size() >= 20) {
            throw new EasiSqlException("Too mush executeBatch Thread");
        }
        String uuid = UUID.randomUUID().toString();
        DBCommitThread newThread = new DBCommitThread(connection, sqlList, uuid);
        newThread.start();
        dbCommitThreadMap.put(uuid, newThread);
        return uuid;
    }

    // 触发设置指令级别,二段式提交的提交指令
    public static void setCurrentThreadCommitLevel (String threadId, String commitLevel) {
        if (ThreadFactory.dbCommitThreadMap.containsKey(threadId)) {
            DBCommitThread dbCommitThread = ThreadFactory.dbCommitThreadMap.get(threadId);
            dbCommitThread.setCommitLevel(commitLevel);
        }
    }

    // 清理程序定期清理
    public static void clearEndThread () {
        Set<String> mapSet = ThreadFactory.dbCommitThreadMap.keySet();
        Iterator it = mapSet.iterator();
        while(it.hasNext()) {
            DBCommitThread dbCommitThread = (DBCommitThread) it.next();
            if (dbCommitThread.end) {
                it.remove();
            }
        }
    }
}

具体的业务流程对应上述的方法如下

  1. createDBCommitThread开始创建
  2. notifyToServer通知中央管理层更新变更状态和数据
  3. 等待指令触发setCurrentThreadCommitLevel,分别对应提交成功和提交回滚
  4. clearEndThread清理程序运行
  5. 中央控制程序关闭数据变更

结语

当然,整体的软件架构远远没有描述的这么简单,平台级别的软件体系涉及到了非常庞大的数据库个数,其中对数据执行层、管理层、用户验证层做了分布式的拆分,数据执行层参照微服务里面的sidecar设计,它是数据库的一个配套服务,执行层配合管理层做了服务发现,进行机器的统一管理,多活,对外公布API服务,死机器踢出,而数据变更自动化仅仅是数据中台设计中的一个子模块设计,究其本质,都是为了让应用的开发维护能更方便,可以给用户提供更优质、快速的数据服务。

顶层的代码设计,Hunter为管理层,watch dog为执行层,双方均支持可扩展
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 194,242评论 5 459
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 81,769评论 2 371
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 141,484评论 0 319
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 52,133评论 1 263
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 61,007评论 4 355
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 46,080评论 1 272
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 36,496评论 3 381
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 35,190评论 0 253
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 39,464评论 1 290
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,549评论 2 309
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,330评论 1 326
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 32,205评论 3 312
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 37,567评论 3 298
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,889评论 0 17
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,160评论 1 250
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 41,475评论 2 341
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 40,650评论 2 335

推荐阅读更多精彩内容