【金仓数据库征文】MyBatis+Flowable双剑合璧——金仓开发实战之工作流引擎适配全记录

前言

做了数据库折磨多年，天天跟SQL打交道，自以为对数据库那点事门儿清。直到今年接了个企业审批系统的信创改造项目，才发现自己想简单了。这个项目用的是Flowable工作流引擎，审批流全跑在数据库上，表就有几十张，SQL更是成百上千条。甲方要求从MySQL切到金仓KES V9R3C18 MySQL兼容版，工期还紧。

一开始我觉得不就是换个驱动改个URL嘛，半天的事。真干起来才发现，坑比想象的多太多了。MyBatis这边还好说，Flowable那边直接连不上——它默认支持的数据库列表里根本没有金仓。

这篇文章就把当时踩的坑、摸的路、最后怎么搞定的，从头到尾记录下来。从最基础的Java驱动连接，到MyBatis集成优化，再到Flowable工作流引擎的方言适配和全流程验证，全部是项目截图和源代码的形式给出，详细说明。

一、Java连接金仓——驱动与连接基础

1.1 驱动包与依赖

先从最基础的说起。金仓官方JDBC驱动是 kingbase8-jdbc，Maven坐标：

<dependency>
    <groupId>cn.com.kingbase</groupId>
    <artifactId>kingbase8-jdbc</artifactId>
    <version>8.6.0</version>
</dependency>

这个驱动是基于PostgreSQL JDBC改的，所以接口层面和PG驱动高度一致，但针对MySQL兼容模式做了不少适配。实际用下来，JDBC4.2的标准接口基本都支持，Connection、Statement、PreparedStatement、ResultSet 这些核心接口行为正常。

1.2 原生JDBC连接示例

每次做新库适配，我习惯先用最朴素的JDBC跑通，确认底层没问题，再往上接框架。这样出问题好定位。

import java.sql.*;

public class KingbaseJdbcTest {
    
    private static final String DRIVER = "com.kingbase8.Driver";
    private static final String URL = "jdbc:kingbase8://101.33.212.149:54321/test?useUnicode=true&characterEncoding=utf8";
    private static final String USER = "system";
    private static final String PASSWORD = "your_password";
    
    public static void main(String[] args) {
        Connection conn = null;
        try {
            Class.forName(DRIVER);
            conn = DriverManager.getConnection(URL, USER, PASSWORD);
            
            if (conn != null && !conn.isClosed()) {
                System.out.println("=== 金仓数据库连接成功 ===");
                
                // 1. 查询数据库版本
                Statement stmt = conn.createStatement();
                ResultSet rs = stmt.executeQuery("SELECT version()");
                if (rs.next()) {
                    System.out.println("数据库版本: " + rs.getString(1).substring(0, 50) + "...");
                }
                
                // 2. 查询当前用户和Schema
                rs = stmt.executeQuery("SELECT current_user, current_schema()");
                if (rs.next()) {
                    System.out.println("当前用户: " + rs.getString(1));
                    System.out.println("当前Schema: " + rs.getString(2));
                }
                
                // 3. 验证BLOB/CLOB支持
                rs = stmt.executeQuery(
                    "SELECT 'CLOB测试文本'::TEXT AS clob_col, " +
                    "E'\\\\x48454C4C4F'::bytea AS blob_col");
                if (rs.next()) {
                    System.out.println("CLOB读取: " + rs.getString("clob_col"));
                    byte[] blob = rs.getBytes("blob_col");
                    System.out.println("BLOB读取: " + new String(blob));
                }
                
                // 4. 验证事务支持
                conn.setAutoCommit(false);
                stmt.execute("CREATE TABLE IF NOT EXISTS t_conn_test (id INT, name VARCHAR(50))");
                stmt.execute("INSERT INTO t_conn_test VALUES (1, 'test')");
                conn.rollback();  // 回滚测试
                
                // 验证回滚后数据不存在
                rs = stmt.executeQuery("SELECT count(*) FROM t_conn_test");
                if (rs.next()) {
                    System.out.println("事务回滚验证: count=" + rs.getInt(1) + "（应为0）");
                }
                
                stmt.execute("DROP TABLE IF EXISTS t_conn_test");
                
                rs.close();
                stmt.close();
            }
        } catch (ClassNotFoundException e) {
            System.err.println("驱动加载失败: " + e.getMessage());
        } catch (SQLException e) {
            System.err.println("SQL错误: " + e.getMessage());
            System.err.println("错误码: " + e.getSQLState());
            System.err.println("错误位置: " + e.getErrorCode());
        } finally {
            if (conn != null) {
                try { conn.close(); } catch (SQLException e) { e.printStackTrace(); }
            }
        }
    }
}

这段代码别看它简单，每次新环境适配我都跑一遍。它能把连接、版本、用户权限、大字段、事务这些最基础的东西一次性验了。哪一步报错，直接定位问题。

1.3 连接URL参数备忘

几个常用的URL参数给大家列一下，省得去翻文档：

reWriteBatchedInserts=true 这个参数后面讲MyBatis批量操作的时候还会提到，对批量插入性能提升很大，建议打开。

二、MyBatis集成金仓最佳实践

2.1 基础配置

SpringBoot + MyBatis 的标准配置，把数据源换成金仓就行：

spring:
  datasource:
    driver-class-name: com.kingbase8.Driver
    url: jdbc:kingbase8://`127.0.0.1`:54321/test?useUnicode=true&characterEncoding=utf8&currentSchema=flow_schema&reWriteBatchedInserts=true
    username: system
    password: your_password
    type: com.zaxxer.hikari.HikariDataSource
    hikari:
      maximum-pool-size: 30
      minimum-idle: 10
      connection-timeout: 30000
      max-lifetime: 1800000
      idle-timeout: 600000
      connection-test-query: SELECT 1
      pool-name: KingbaseFlowPool

mybatis:
  mapper-locations: classpath:mapper/**/*.xml
  type-aliases-package: com.approval.entity
  configuration:
    map-underscore-to-camel-case: true
    cache-enabled: false
    log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl

这里重点说一下 currentSchema=flow_schema。MySQL是多库架构，不同业务建不同的database。金仓是PG架构，一个database下面建多个schema。业务表放单独的schema里，既便于权限管理，又不会跟系统表混在一起。

2.2 分页插件适配

MyBatis的分页插件PageHelper，用的时候注意方言。金仓的分页语法和PG一样，是 LIMIT x OFFSET y，所以方言配置成 PostgreSQL 就行：

<!-- PageHelper 配置 -->
<dependency>
    <groupId>com.github.pagehelper</groupId>
    <artifactId>pagehelper-spring-boot-starter</artifactId>
    <version>1.4.6</version>
</dependency>

pagehelper:
  helper-dialect: postgresql  # 金仓用postgreSQL方言
  reasonable: true
  support-methods-arguments: true
  params: count=countSql

helper-dialect 配 postgresql 就能正常工作。我当时还担心会不会有问题，测了各种分页场景——单表、多表关联、排序、分组，都没问题。

2.3 TypeHandler 适配

数组类型

金仓支持数组类型，比如 VARCHAR[]、INT[]。业务里有时候要存标签、角色列表这类东西，数组类型比JSON更轻量。

自定义一个List转数组的TypeHandler：

@MappedJdbcTypes(JdbcType.ARRAY)
@MappedTypes(List.class)
public class StringArrayTypeHandler extends BaseTypeHandler<List<String>> {

    @Override
    public void setNonNullParameter(PreparedStatement ps, int i, 
            List<String> parameter, JdbcType jdbcType) throws SQLException {
        Connection conn = ps.getConnection();
        Array array = conn.createArrayOf("VARCHAR", parameter.toArray());
        ps.setArray(i, array);
    }

    @Override
    public List<String> getNullableResult(ResultSet rs, String columnName) 
            throws SQLException {
        Array array = rs.getArray(columnName);
        return array == null ? null : Arrays.asList((String[]) array.getArray());
    }

    @Override
    public List<String> getNullableResult(ResultSet rs, int columnIndex) 
            throws SQLException {
        Array array = rs.getArray(columnIndex);
        return array == null ? null : Arrays.asList((String[]) array.getArray());
    }

    @Override
    public List<String> getNullableResult(CallableStatement cs, int columnIndex) 
            throws SQLException {
        Array array = cs.getArray(columnIndex);
        return array == null ? null : Arrays.asList((String[]) array.getArray());
    }
}

Mapper XML里用的时候指定 typeHandler：

<resultMap id="UserMap" type="com.approval.entity.SysUser">
    <id column="id" property="id"/>
    <result column="username" property="username"/>
    <result column="roles" property="roles" 
            typeHandler="com.approval.handler.StringArrayTypeHandler"/>
</resultMap>

JSONB类型

JSONB字段映射成String最简单：

<result column="ext_info" property="extInfo" jdbcType="OTHER"/>

如果想直接映射成对象，自己写个 Jackson 的 TypeHandler 也不难。但我一般建议业务层做序列化反序列化，SQL层就传字符串，简单直接，出问题也好排查。

2.4 批量操作优化

批量插入是项目里最常见的性能瓶颈。几种方式对比一下：

方式一：foreach拼接（最常用）

<insert id="batchInsert">
    INSERT INTO sys_approval_log (task_id, operator, action, content, create_time)
    VALUES
    <foreach collection="list" item="item" separator=",">
        (#{item.taskId}, #{item.operator}, #{item.action}, 
         #{item.content}, #{item.createTime})
    </foreach>
</insert>

这种方式简单直接，一次插几百条没问题。但要注意，金仓对SQL长度有限制，单次插入数据量太大（比如几千条以上）可能会爆。

方式二：BATCH模式（推荐，量大用这个）

MyBatis的ExecutorType设为BATCH，配合JDBC批量更新：

@Service
public class BatchService {

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Transactional
    public void batchInsertLogs(List<ApprovalLog> logs) {
        // 打开BATCH模式的Session
        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            ApprovalLogMapper mapper = session.getMapper(ApprovalLogMapper.class);
            
            int count = 0;
            for (ApprovalLog log : logs) {
                mapper.insert(log);
                count++;
                // 每1000条提交一次，防止内存溢出
                if (count % 1000 == 0) {
                    session.commit();
                    session.clearCache();
                }
            }
            session.commit();
        }
    }
}

BATCH模式下，MyBatis会把SQL攒到一起发给数据库，比foreach性能还好一些。而且不会因为数据量太大导致SQL过长。

提醒：URL里的 reWriteBatchedInserts=true 一定要开，这个参数会把批量插入重写成多值INSERT，性能提升明显。我当时测了一下，开了之后批量插入速度快了3倍左右。

2.5 动态SQL注意事项

MyBatis的动态SQL在金仓上基本都能用，但有几个地方要留意：

1. limit 参数

MySQL里可以写 LIMIT #{offset}, #{size}，金仓（PG风格）里是 LIMIT #{size} OFFSET #{offset}。因为我们用了PageHelper分页插件，这个一般不用自己写。但手写分页SQL的时候别搞反了。

2. 字符串拼接

还是那个老问题，|| 是逻辑或不是拼接。MyBatis的XML里如果写了拼接SQL，记得用 CONCAT：

<!-- 错误写法，MySQL兼容模式下有坑 -->
<if test="keyword != null and keyword != ''">
    AND username LIKE '%' || #{keyword} || '%'
</if>

<!-- 正确写法 -->
<if test="keyword != null and keyword != ''">
    AND username LIKE CONCAT('%', #{keyword}, '%')
</if>

3. ON DUPLICATE KEY UPDATE

MySQL特有的upsert语法，金仓不完全支持。改成PG风格的 ON CONFLICT：

<insert id="upsertUser">
    INSERT INTO sys_user (username, real_name, email, update_time)
    VALUES (#{username}, #{realName}, #{email}, NOW())
    ON CONFLICT (username)
    DO UPDATE SET
        real_name = EXCLUDED.real_name,
        email = EXCLUDED.email,
        update_time = NOW()
</insert>

EXCLUDED 这个关键字很好用，代表"本来要插入的那一行"，更新的时候直接引用就行。

三、Flowable工作流引擎适配金仓

这部分是重头戏。Flowable官方支持的数据库有MySQL、PostgreSQL、Oracle、SQL Server、DB2、H2等，但就是没有金仓。那能不能用？答案是能，但得自己适配。

3.1 适配思路

Flowable判断数据库类型是根据连接的元数据来的。它拿 DatabaseMetaData.getDatabaseProductName() 的返回值做匹配，如果是"MySQL"就走MySQL逻辑，是"PostgreSQL"就走PG逻辑。

金仓的驱动返回的 productName 是 “KingbaseES”，Flowable不认识，直接报错：

org.flowable.common.engine.api.FlowableException: 
could not update engine tables: no database type defined for 'KingbaseES'

解决思路有两个：

方案一：把金仓伪装成PostgreSQL

金仓是基于PG的，语法高度兼容。如果能让Flowable以为自己连的是PG，那它就会用PG的SQL和建表语句，大概率能跑通。

方案二：自定义金仓数据库类型

扩展Flowable源码，加一个KingbaseES的数据库类型和对应的SQL模板。这个更规范，但改源码维护成本高。

我选了方案一，理由很简单：稳、快、改动小，出了问题好回退。项目工期紧的时候，能用最少的改动把事办了，才是合格的工程师。

3.2 具体实现

具体怎么伪装？两个步骤。

第一步：自定义数据源包装类

写个包装类，把 DatabaseMetaData 里的数据库产品名改成 “PostgreSQL”：

public class KingbaseToPostgresDataSource implements DataSource {

    private final DataSource delegate;

    public KingbaseToPostgresDataSource(DataSource delegate) {
        this.delegate = delegate;
    }

    @Override
    public Connection getConnection() throws SQLException {
        Connection conn = delegate.getConnection();
        // 返回包装后的Connection，伪装成PostgreSQL
        return new KingbaseConnectionWrapper(conn);
    }

    @Override
    public Connection getConnection(String username, String password) 
            throws SQLException {
        Connection conn = delegate.getConnection(username, password);
        return new KingbaseConnectionWrapper(conn);
    }

    // 其他方法全部委托给delegate
    @Override public PrintWriter getLogWriter() { return delegate.getLogWriter(); }
    @Override public void setLogWriter(PrintWriter out) { delegate.setLogWriter(out); }
    @Override public void setLoginTimeout(int seconds) { delegate.setLoginTimeout(seconds); }
    @Override public int getLoginTimeout() { return delegate.getLoginTimeout(); }
    @Override public Logger getParentLogger() { return delegate.getParentLogger(); }
    @Override public <T> T unwrap(Class<T> iface) { return delegate.unwrap(iface); }
    @Override public boolean isWrapperFor(Class<?> iface) { return delegate.isWrapperFor(iface); }
}

然后是Connection的包装类，重点在 getMetaData()：

public class KingbaseConnectionWrapper implements Connection {

    private final Connection delegate;

    public KingbaseConnectionWrapper(Connection delegate) {
        this.delegate = delegate;
    }

    @Override
    public DatabaseMetaData getMetaData() throws SQLException {
        DatabaseMetaData original = delegate.getMetaData();
        // 返回包装后的MetaData，修改数据库产品名称
        return new KingbaseMetaDataWrapper(original, delegate);
    }

    // 其他方法全部委托
    @Override public Statement createStatement() { return delegate.createStatement(); }
    @Override public PreparedStatement prepareStatement(String sql) { return delegate.prepareStatement(sql); }
    // ... 省略其他几十个委托方法，全部直接调用delegate对应的方法
}

最后是MetaData包装类，只改产品名这一个地方：

public class KingbaseMetaDataWrapper implements DatabaseMetaData {

    private final DatabaseMetaData delegate;
    private final Connection connection;

    public KingbaseMetaDataWrapper(DatabaseMetaData delegate, Connection connection) {
        this.delegate = delegate;
        this.connection = connection;
    }

    @Override
    public String getDatabaseProductName() throws SQLException {
        // 关键！返回PostgreSQL，让Flowable以为连的是PG
        return "PostgreSQL";
    }
    
    @Override
    public String getDatabaseProductVersion() throws SQLException {
        return delegate.getDatabaseProductVersion();
    }
    
    @Override
    public Connection getConnection() throws SQLException {
        // 返回包装后的Connection，保持一致
        return connection;
    }

    // 其他方法全部委托给delegate
    @Override public boolean supportsTransactions() { return delegate.supportsTransactions(); }
    @Override public boolean supportsBatchUpdates() { return delegate.supportsBatchUpdates(); }
    @Override public ResultSet getTables(String catalog, String schemaPattern, 
            String tableNamePattern, String[] types) { 
        return delegate.getTables(catalog, schemaPattern, tableNamePattern, types); 
    }
    // ... 还有几十个方法，全部委托就行
}

第二步：SpringBoot配置

把自定义的数据源包装一下，注入到Flowable的引擎配置里：

@Configuration
public class FlowableConfig {

    @Bean
    @Primary
    public DataSource kingbaseDataSource(DataSourceProperties properties) {
        HikariDataSource original = properties.initializeDataSourceBuilder()
                .type(HikariDataSource.class).build();
        // 包装一层，让Flowable以为是PostgreSQL
        return new KingbaseToPostgresDataSource(original);
    }

    @Bean
    public EngineConfigurationConfigurer<SpringProcessEngineConfiguration> 
            processEngineConfigurer() {
        return configuration -> {
            // 使用PostgreSQL的数据库类型
            configuration.setDatabaseType("postgres");
            // 不自动创建/更新表，首次初始化后建议关闭
            configuration.setDatabaseSchemaUpdate("true");
            // 历史级别
            configuration.setHistoryLevel(HistoryLevel.AUDIT);
            // 自定义UUID生成器
            configuration.setIdGenerator(new StrongUuidGenerator());
        };
    }
}

configuration.setDatabaseType("postgres") 这行也很关键，它告诉Flowable用PostgreSQL的SQL模板，不是自动检测。双保险。

3.3 启动验证

配置好之后直接启动SpringBoot应用，如果没报错就说明适配成功了。

验证一下表有没有创建成功：

-- 查Flowable表数量（应该有40多张表）
SELECT count(*) 
FROM information_schema.tables 
WHERE table_schema = 'flow_schema' 
  AND table_name LIKE 'act_%';

-- 结果大概是 48 张表左右

再跑一个最简单的流程实例验证一下：

@Service
public class FlowTestService {

    @Autowired
    private RuntimeService runtimeService;
    
    @Autowired
    private TaskService taskService;
    
    @Autowired
    private HistoryService historyService;

    public void testSimpleProcess() {
        // 1. 启动流程
        ProcessInstance pi = runtimeService.startProcessInstanceByKey("simpleApproval");
        System.out.println("流程实例ID: " + pi.getId());
        System.out.println("流程定义ID: " + pi.getProcessDefinitionId());
        
        // 2. 查询待办任务
        List<Task> tasks = taskService.createTaskQuery()
                .processInstanceId(pi.getId())
                .list();
        System.out.println("待办任务数: " + tasks.size());
        
        // 3. 完成第一个任务
        if (!tasks.isEmpty()) {
            Task task = tasks.get(0);
            Map<String, Object> variables = new HashMap<>();
            variables.put("approved", true);
            taskService.complete(task.getId(), variables);
            System.out.println("完成任务: " + task.getName());
        }
        
        // 4. 查询历史
        long count = historyService.createHistoricTaskInstanceQuery()
                .processInstanceId(pi.getId())
                .finished()
                .count();
        System.out.println("已完成历史任务数: " + count);
    }
}

这段代码能跑通，说明Flowable在金仓上基本没问题了。我当时测了启动流程、完成任务、驳回、转办、会签、子流程、定时边界事件，都正常。

3.4 踩过的几个坑

适配过程中踩了几个坑，这里记下来，大家少走弯路。

坑一：主键生成策略

Flowable默认用的是自己的UUID生成器（StrongUuidGenerator），这个没问题，不是数据库自增的，不涉及兼容问题。但如果你配置了数据库序列生成主键，就要注意一下，金仓的序列语法和PG一样，是 nextval('seq_name')，这个没问题。

坑二：锁表死锁

Flowable有自己的乐观锁和悲观锁机制。在金仓上跑的时候，发现高并发下偶尔会死锁。查了一下，是行锁等待超时。

解决办法：调大 lockWaitTime 或者优化业务并发。

configuration.setLockWaitTime(60000); // 锁等待时间，默认5000ms

我这边的项目并发不高，调到60秒基本就没再出现过。如果是高并发场景，可能还要进一步优化。

坑三：JSON变量存储

Flowable的流程变量支持各种类型。存JSON格式的变量的时候，注意类型别乱。我这边的做法是统一存字符串，业务层自己做序列化反序列化，最稳。

// 存JSON变量
ObjectMapper mapper = new ObjectMapper();
String jsonStr = mapper.writeValueAsString(businessData);
runtimeService.setVariable(executionId, "businessData", jsonStr);

// 取JSON变量
String jsonStr = (String) runtimeService.getVariable(executionId, "businessData");
BusinessData data = mapper.readValue(jsonStr, BusinessData.class);

别用复杂对象直接存，序列化方式变了容易出问题。字符串是最保险的。

坑四：定时任务执行

Flowable自带定时任务执行器（AsyncExecutor），用来处理定时器、异步任务这些。在金仓上跑的时候，定时任务有时候会卡住。

查了一下原因，是定时任务在取任务的时候，用了 SELECT ... FOR UPDATE 加锁，金仓的行锁行为和PG有些细微差异。

解决办法：配置一下 asyncExecutor 的参数，降低并发：

configuration.setAsyncExecutorActivate(true);
configuration.setAsyncExecutorCorePoolSize(4);   // 核心线程数，别太大
configuration.setAsyncExecutorMaxPoolSize(8);    // 最大线程数
configuration.setAsyncExecutorQueueSize(100);    // 队列大小

我这边从默认的10/50/256 降到 4/8/100 之后，定时任务就稳定了。毕竟审批系统的定时任务量不大，不需要那么多线程。

3.5 上线后的稳定性

项目上线到现在跑了两个多月，处理了五千多个审批流程，大的问题没有。数据库层面除了有几条慢SQL优化了一下索引，整体很稳。

Flowable引擎层没有出过兼容性问题。事实证明，把金仓伪装成PostgreSQL给Flowable用，这条路是走得通的。毕竟金仓底层就是PG架构，语法高度兼容，ORM和工作流引擎只要认PG方言，基本都能跑。

四、开发大赛参赛感想

用过金仓数据库之后才发现其实没有那么困难，其实就是因为做这个Flowable适配项目做出来的信心。想着自己既然把工作流引擎都跑通了，不如整理整理去参赛，顺便跟大家交流一下。

参加比赛最大的收获，倒不是拿了什么奖，而是逼着自己把零散的经验系统化了。平时做项目，遇到问题解决了就完事，写参赛材料的时候得从头到尾理清楚，从架构设计到具体实现，从问题分析到解决方案，得有完整的逻辑链。这个梳理的过程，比什么都有价值。

另外就是通过比赛认识了一批同样在做信创的同行。平时大家都在各自的项目里闷头干，交流不多。比赛的时候凑到一起一聊，发现你踩过的坑我也踩过，你摸索出来的路子我也试过，那种共鸣感特别强。国产数据库的生态还在建设中，同行之间多交流少重复踩坑，比什么都重要。

最深的一个感想是：信创这条路，不能光靠数据库厂商自己往前冲，得有大量的开发者、集成商、最终用户一起参与进来。数据库好不好用，不是厂商说了算，是开发者写代码的时候说了算，是运维人员半夜排查问题的时候说了算。每一个适配方案、每一次踩坑记录、每一篇技术文章，都是在给国产数据库的生态添砖加瓦。

作为一个十几年的老DBA，以前我对国产数据库是持怀疑态度的，总觉得差得远。这两年扎扎实实做了几个项目，用了金仓之后，观念真的变了。不是说已经完美了——差距肯定还是有的——但这个差距正在快速缩小。在很多业务场景下，国产数据库已经完全能顶得住了。

技术在进步，人也得进步。守着老一套Oracle、MySQL的经验吃老本，这条路会越来越窄。主动去了解、去使用、去贡献，跟着国产数据库一起成长，才是我们这代技术人该有的态度。

后面我打算继续把这套Flowable适配方案完善一下，整理成一个开源的starter，让更多人不用再从零开始踩坑。也算是为信创生态出一份力吧。