Fescar - RM InsertExecutor介绍

简介: 开篇 这篇文章的目的是讲解RM Executor模块当中一些通用的方法,这些方法在各个Executor的父类当中实现的,各个子类Executor模块都会复用,因此抽取出来统一的进行讲解。 个人是认为抽取通用的内容放在一篇文章讲解完后可以针对每类Executor讲解特有的功能,这样能够有更好的理解。

开篇

 这篇文章的目的是讲解RM Executor模块当中一些通用的方法,这些方法在各个Executor的父类当中实现的,各个子类Executor模块都会复用,因此抽取出来统一的进行讲解。

 个人是认为抽取通用的内容放在一篇文章讲解完后可以针对每类Executor讲解特有的功能,这样能够有更好的理解。这篇文章讲解Executor的实现类InsertExecutor。


类依赖图


说明:

  • 着重讲解InsertExecutor实现类。


InsertExecutor方法介绍

public class InsertExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
 
   protected TableRecords beforeImage() throws SQLException {
        return TableRecords.empty(getTableMeta());
    }

}

public class TableRecords {

    public static TableRecords empty(TableMeta tableMeta) {
        return new TableRecords(tableMeta) {
            @Override
            public int size() {
                return 0;
            }

            @Override
            public List<Field> pkRows() {
                return new ArrayList<>();
            }

            @Override
            public void add(Row row) {
                throw new UnsupportedOperationException("xxx");
            }

            @Override
            public TableMeta getTableMeta() {
                throw new UnsupportedOperationException("xxx");
            }
        };
    }
}

说明:

  • Insert操作的执行前镜像顾名思义是空的,所以beforeImage()方法返回空记录
  • TableRecords.empty返回的是一个没有记录的TableRecords对象。


public class InsertExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {

    protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
        SQLInsertRecognizer recogizier = (SQLInsertRecognizer)sqlRecognizer;
        List<String> insertColumns = recogizier.getInsertColumns();
        TableMeta tmeta = getTableMeta();
        TableRecords afterImage = null;
        if (tmeta.containsPK(insertColumns)) {
            // 处理插入数据包含主键的情况
            List<Object> pkValues = null;
            String pk = tmeta.getPkName();
            for (int paramIdx = 0; paramIdx < insertColumns.size(); paramIdx++) {
                if (insertColumns.get(paramIdx).equalsIgnoreCase(pk)) {
                    // 处理PreparedStatement类型,从参数列表中提取
                    if (statementProxy instanceof PreparedStatementProxy) {
                        pkValues = ((PreparedStatementProxy) statementProxy).getParamsByIndex(paramIdx);
                    } else {
                        // 处理Statement类型,直接从SQL语句中提取
                        List<List<Object>> insertRows = recogizier.getInsertRows();
                        pkValues = new ArrayList<>(insertRows.size());
                        
                        for (List<Object> row : insertRows) {
                            pkValues.add(row.get(paramIdx));
                        }
                    }

                    //主键只有一个
                    break;
                }
            }
            if (pkValues == null) {
                throw new ShouldNeverHappenException();
            }
            afterImage = getTableRecords(pkValues);

        } else {
            // 处理主键自动生成的场景,也就是不包含主键的情况。
            Map<String, ColumnMeta> pkMetaMap = getTableMeta().getPrimaryKeyMap();
            if (pkMetaMap.size() != 1) {
                throw new NotSupportYetException();
            }

            // 获取primary key的元数据信息
            ColumnMeta pkMeta = pkMetaMap.values().iterator().next();
            if (!pkMeta.isAutoincrement()) {
                throw new ShouldNeverHappenException();
            }

            // 获取generatedKeys()获取生成的主键
            ResultSet genKeys = null;
            try {
                genKeys = statementProxy.getTargetStatement().getGeneratedKeys();
            } catch (SQLException e) {
               
                if ("S1009".equalsIgnoreCase(e.getSQLState())) {
                    genKeys = statementProxy.getTargetStatement().executeQuery("SELECT LAST_INSERT_ID()");
                } else {
                    throw e;
                }
            }
            List<Object> pkValues = new ArrayList<>();
            while (genKeys.next()) {
                Object v = genKeys.getObject(1);
                pkValues.add(v);
            }

            // 生成执行后镜像
            afterImage = getTableRecords(pkValues);

        }

        if (afterImage == null) {
            throw new SQLException("Failed to build after-image for insert");
        }

        return afterImage;
    }
}

说明:

  • afterImage准备执行后的镜像,针对Insert操作就是新增的记录。
  • afterImage处理分为两种场景,分为插入记录包含主键字段和不包含主键字段
  • 记录包含主键字段场景下,区分preparedStatement和Statement两种场景,前者去传入参数中获取,后者从执行的SQL当中解析。
  • 不包含主键字段场景下,通过获取getGeneratedKeys()返回的值获取。
  • 获取主键的值以后pkValues,然后通过getTableRecords()生成执行后镜像


public class InsertExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {

    private TableRecords getTableRecords(List<Object> pkValues) throws SQLException {
        TableRecords afterImage;
        String pk = getTableMeta().getPkName();
        StringBuffer selectSQLAppender = new StringBuffer("SELECT * FROM " + getTableMeta().getTableName() + " WHERE ");
        for (int i = 1; i <= pkValues.size(); i++) {
            selectSQLAppender.append(pk + "=?");
            if (i < pkValues.size()) {
                selectSQLAppender.append(" OR ");
            }
        }
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            ps = statementProxy.getConnection().prepareStatement(selectSQLAppender.toString());

            for (int i = 1; i <= pkValues.size(); i++) {
                ps.setObject(i, pkValues.get(i - 1));
            }

            rs = ps.executeQuery();
            afterImage = TableRecords.buildRecords(getTableMeta(), rs);

        } finally {
            if (rs != null) {
                rs.close();
            }
            if (ps != null) {
                ps.close();
            }
        }
        return afterImage;
    }
}

说明:

  • getTableRecords通过针对主键信息拼接SELECT的SQL语句拼接成根据主键查询字段信息的SQL
  • 通过buildRecords()方法针对查询结果ResultSet进行处理。
  • 生成执行后镜像信息在buildRecords()内部实现。


public class TableRecords {

    public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) throws SQLException {
        TableRecords records = new TableRecords(tmeta);
        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
        int columnCount = resultSetMetaData.getColumnCount();

        while (resultSet.next()) {
            List<Field> fields = new ArrayList<>(columnCount);
            for (int i = 1; i <= columnCount; i++) {
                String colName = resultSetMetaData.getColumnName(i);
                ColumnMeta col = tmeta.getColumnMeta(colName);
                Field field = new Field();
                field.setName(col.getColumnName());
                if (tmeta.getPkName().equals(field.getName())) {
                    field.setKeyType(KeyType.PrimaryKey);
                }
                field.setType(col.getDataType());
                field.setValue(resultSet.getObject(i));
                fields.add(field);
            }

            Row row = new Row();
            row.setFields(fields);

            records.add(row);
        }
        return records;
    }
}

说明:

  • buildRecords内部遍历查询的结果凭借Row对象。
  • 插入的数据可能有多条,外层while循环获取所有的行数,内层for循环遍历所有的列数。


期待

InsertExecutor生成执行前后镜像的过程分析完后,会接着分析DeleteExecutor。额外的收货就是分析完对于SQL或者说对阿里开源的druid有了一些深入研究的兴趣。

目录
相关文章
|
6月前
|
Java 微服务 Spring
Seata 客户端需要同时启动 TM 和 RM 吗?
Seata 客户端需要同时启动 TM 和 RM 吗?
RM在seata AT模式中如何实现分支事务提交或回滚
RM在seata AT模式中如何实现分支事务提交或回滚
462 0
|
SQL 中间件 FESCAR
分布式事务中间件 Fescar—RM 模块源码解读
前言 在SOA、微服务架构流行的年代,许多复杂业务上需要支持多资源占用场景,而在分布式系统中因为某个资源不足而导致其它资源占用回滚的系统设计一直是个难点。我所在的团队也遇到了这个问题,为解决这个问题上,团队采用的是阿里开源的分布式中间件Fescar的解决方案,并详细了解了Fescar内部的工作原理,解决在使用Fescar中间件过程中的一些疑虑的地方,也为后续团队在继续使用该中间件奠定理论基础。
27638 71
|
XML Java 中间件
Seata 分支事务
前面,我们已经介绍了 Seata 的整体设计思想,接下来我们深入到其实现细节中,本文先来介绍 Seata 中分支事务的整体实现思想。
|
Java 微服务 Spring
Seata 客户端需要同时启动 RM 和 TM 吗?
在分析启动部分源码时,我发现 GlobalTransactionScanner 会同时启动 RM 和 TM client,但根据 Seata 的设计来看,TM 负责全局事务的操作,如果一个服务中不需要开启全局事务,此时是不需要启动 TM client的,也就是说项目中如果没有全局事务注解,此时是不是就不需要初始化 TM client 了,因为不是每个微服务,都需要 GlobalTransactional,它此时仅仅作为一个 RM client 而已。
147 0
Seata 客户端需要同时启动 RM 和 TM 吗?
|
Shell Nacos 微服务
seata夸服务器服务注册RM失败问题
seata夸服务器服务注册RM失败问题
637 0
|
SQL 中间件 Java
分布式事务中间件Fescar—RM模块源码解读
在SOA、微服务架构流行的年代,许多复杂业务上需要支持多资源占用场景,而在分布式系统中因为某个资源不足而导致其它资源占用回滚的系统设计一直是个难点。我所有团队也遇到了这个问题,为解决这个问题上,团队采用的是阿里开源的分布式中间件Fescar的解决方案,并详细了解了Fescar内部的工作原理,解决在使用Fescar中间件过程中的一些疑虑的地方,也为后续团队在继续使用该中间件奠定理论基础。
2760 2
|
FESCAR SQL
Fescar - RM SelectForUpdateExecutor介绍
开篇  这篇文章的目的是讲解RM Executor模块当中一些通用的方法,这些方法在各个Executor的父类当中实现的,各个子类Executor模块都会复用,因此抽取出来统一的进行讲解。  个人是认为抽取通用的内容放在一篇文章讲解完后可以针对每类Executor讲解特有的功能,这样能够有更好的理解。
1180 0
|
SQL FESCAR
Fescar - RM DeleteExecutor介绍
开篇  这篇文章的目的是讲解RM Executor模块当中一些通用的方法,这些方法在各个Executor的父类当中实现的,各个子类Executor模块都会复用,因此抽取出来统一的进行讲解。  个人是认为抽取通用的内容放在一篇文章讲解完后可以针对每类Executor讲解特有的功能,这样能够有更好的理解。
1090 0
|
SQL FESCAR 安全
Fescar - RM UpdateExecutor介绍
开篇  这篇文章的目的是讲解RM Executor模块当中一些通用的方法,这些方法在各个Executor的父类当中实现的,各个子类Executor模块都会复用,因此抽取出来统一的进行讲解。  个人是认为抽取通用的内容放在一篇文章讲解完后可以针对每类Executor讲解特有的功能,这样能够有更好的理解。
1120 0