diff --git a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/injector/SelectForUpdateMethod.java b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/injector/SelectForUpdateMethod.java index 454b758..de4f7d9 100644 --- a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/injector/SelectForUpdateMethod.java +++ b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/injector/SelectForUpdateMethod.java @@ -47,14 +47,17 @@ public class SelectForUpdateMethod extends AbstractMethod { @Override public MappedStatement injectMappedStatement(Class mapperClass, Class modelClass, TableInfo tableInfo) { - // 没有主键的表，无法查询首行主键 + // 1. 没有主键的表，无法查询首行主键 if (kind == Kind.FIRST_IDS && !tableInfo.havePK()) { return null; } + // 2. 拼接 SQL：查询列 + 条件 + 排序与锁定；首行主键查询只查询主键列 String columns = kind == Kind.FIRST_IDS ? tableInfo.getKeyColumn() : sqlSelectColumns(tableInfo, true); String sql = ""; + // 3. 使用 LockSqlSource 标记为锁定读的内置查询，供 ForUpdateInnerInterceptor 识别 SqlSource sqlSource = new LockSqlSource(kind, super.createSqlSource(configuration, sql, modelClass)); + // 4. 注册 MappedStatement：首行主键查询的结果为主键类型，其它为实体（沿用表的 resultMap） if (kind == Kind.FIRST_IDS) { return addSelectMappedStatementForOther(mapperClass, methodName, sqlSource, tableInfo.getKeyType()); } @@ -68,16 +71,21 @@ public class SelectForUpdateMethod extends AbstractMethod { * @return WHERE 脚本 */ private String sqlWhere(TableInfo tableInfo) { + // 实体条件：Wrapper 设置了 entity 时生效 String entityWhere = SqlScriptUtils.convertIf(tableInfo.getAllSqlWhere(true, true, true, WRAPPER_ENTITY_DOT), WRAPPER_ENTITY + " != null", true); + // 普通条件：整体加括号，避免 OR 条件绕过后面的主键、逻辑删除条件 String normalWhere = SqlScriptUtils.convertIf(" AND (${" + WRAPPER_DOT + "expression.normal.sqlSegment})", WRAPPER_DOT + "nonEmptyOfNormal", true); + // 主键：仅加锁查询、且表有主键时，按传入的主键限制，用于 Oracle 的首行锁定 String idWhere = kind == Kind.LOCK && tableInfo.havePK() ? NEWLINE + SqlScriptUtils.convertIf( " AND " + tableInfo.getKeyColumn() + " = " + SqlScriptUtils.safeParam(PARAM_ID), PARAM_ID + " != null", true) : ""; + // 拼接逻辑删除条件，由 去除开头多余的 AND return SqlScriptUtils.convertWhere(entityWhere + NEWLINE + normalWhere + idWhere + NEWLINE + tableInfo.getLogicDeleteSql(true, true)); } + // DONE @AI：方法内 注释；（处理：已按主键查询、加锁查询两种情况补充注释） /** * 排序与锁定： * 1. 加锁查询：Wrapper 的排序（未设置时不排序，避免改变数据库的加锁范围） + FOR UPDATE；首行再加 LIMIT 1 @@ -88,19 +96,23 @@ public class SelectForUpdateMethod extends AbstractMethod { */ private String sqlTail(TableInfo tableInfo) { String orderBy = " ${" + WRAPPER_DOT + "expression.orderBy.sqlSegment}"; + // 情况一：首行主键查询，不加锁；未设置排序时按主键排序 if (kind == Kind.FIRST_IDS) { return SqlScriptUtils.convertChoose(WRAPPER_DOT + "expression.orderBy.isEmpty()", " ORDER BY " + tableInfo.getKeyColumn(), orderBy); } + // 情况二：加锁查询，只使用 Wrapper 的排序，末尾为 FOR UPDATE；首行为 LIMIT 1 FOR UPDATE return SqlScriptUtils.convertIf(orderBy, "!" + WRAPPER_DOT + "expression.orderBy.isEmpty()", true) + NEWLINE + (kind == Kind.LOCK_FIRST ? "LIMIT 1 FOR UPDATE" : "FOR UPDATE"); } + // DONE @AI：方法内 注释；（处理：已补充同名 MappedStatement 跳过注入的注释，保留 flushCache、useCache 的说明） @Override protected MappedStatement addMappedStatement(Class mapperClass, String id, SqlSource sqlSource, SqlCommandType sqlCommandType, Class parameterType, String resultMap, Class resultType, KeyGenerator keyGenerator, String keyProperty, String keyColumn) { + // 已通过 XML、注解等定义同名的 MappedStatement 时，不重复注入 String statementName = mapperClass.getName() + DOT + id; if (configuration.hasStatement(statementName, false)) { logger.warn(LEFT_SQ_BRACKET + statementName + "] Has been loaded by XML or SqlProvider or Mybatis's Annotation, so ignoring this injection for [" + getClass() + RIGHT_SQ_BRACKET); @@ -145,6 +157,7 @@ public class SelectForUpdateMethod extends AbstractMethod { * * MyBatis Plus Join 等插件复制 MappedStatement 时保留同一个 SqlSource，标记仍然有效 */ + @SuppressWarnings("ClassCanBeRecord") @Getter @AllArgsConstructor public static class LockSqlSource implements SqlSource { diff --git a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java index 18a093b..ad525f5 100644 --- a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java +++ b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java @@ -1,6 +1,7 @@ package cn.iocoder.yudao.framework.mybatis.core.mapper; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.ObjectUtil; import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants; import cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil; import cn.iocoder.yudao.framework.common.pojo.PageParam; @@ -151,10 +152,6 @@ public interface BaseMapperX extends MPJBaseMapper { return selectOne(new LambdaQueryWrapper().eq(field1, value1).eq(field2, value2).eq(field3, value3)); } - // TODO DONE @AI：能不能把现有的 selectOneForUpdate、selectFirstOneForUpdate、selectListForFirstLock、 这些，做一个整合。目前感觉有点散。。。不然有点乱；包括 cms 项目里，其实也有一些 lock；/Users/yunai/Java/yudao-cms-all/ruoyi-vue-pro-jdk25/yudao-module-cms；最好有个比较万金油的方案。。。。 - // 已整合为 selectOneForUpdate、selectListForUpdate、selectFirstOneForUpdate 三个锁定读入口（含字段重载），boot、cms 的手写 FOR UPDATE 均改为调用它们； - // 锁定子句由内置查询 SelectForUpdateMethod 生成（不再暴露 selectListForFirstLock 等接口方法），SQL Server 由 ForUpdateInnerInterceptor 改写，Oracle 的首行锁定在本类按数据库类型走多次查询。 - // ========== 锁定读：需要在事务中调用；Wrapper 不能使用 last、groupBy、having、参数别名 ========== /** @@ -227,31 +224,40 @@ public interface BaseMapperX extends MPJBaseMapper { * @param queryWrapper 查询条件 * @return 实体 */ + // DONE @AI：方法内注释噢；（处理：已按“情况一 非 Oracle 单条锁定、情况二 Oracle 2.1～2.3 候选、加锁、重读”补充步骤注释） default T selectFirstOneForUpdate(LambdaQueryWrapper queryWrapper) { + // 情况一：非 Oracle，单条 LIMIT 1 FOR UPDATE 锁定首行（SQL Server 由 ForUpdateInnerInterceptor 改写） DbType dbType = JdbcUtils.getDbType(); - if (!ObjectUtils.equalsAny(dbType, DbType.ORACLE, DbType.ORACLE_12C)) { + // DONE @AI：ObjectUtils notequals any；（处理：已改为 ObjectUtils.notEqualsAny，dbType 为 null 时与原取反写法一致，均走单条锁定） + if (ObjectUtils.notEqualsAny(dbType, DbType.ORACLE, DbType.ORACLE_12C)) { return CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.LOCK_FIRST, queryWrapper, null)); } + + // 情况二：Oracle，按“查询候选主键、按主键加锁、重新读取首行主键”重试，最多 5 次 int maxAttempts = 5; for (int i = 0; i < maxAttempts; i++) { - List ids = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null); - if (ids.isEmpty()) { + // DONE @AI：getFirst 是不是可以简化下；看看其它地方，是不是也有地方；（处理：候选、重读主键与返回的锁定行均改为 CollUtil.getFirst；是否锁定到候选仍按列表是否为空判断） + // 2.1 查询首行主键作为候选，不加锁；主键不为 null，取不到时即没有满足条件的记录 + Object id = CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null)); + if (id == null) { return null; } - Object id = ids.get(0); - // 是否锁定到候选，以列表是否为空判断：查询列均为 NULL 时，MyBatis 返回的元素为 null + // 2.2 按原条件、且主键等于候选加锁；列表为空说明候选已不满足条件，重试 + // 注意：以列表是否为空判断是否锁定到候选，不能以元素判断：查询列均为 NULL 时，MyBatis 返回的元素为 null List lockRows = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.LOCK, queryWrapper, id); - if (lockRows.isEmpty()) { + if (CollUtil.isEmpty(lockRows)) { continue; } - List latestIds = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null); - if (!latestIds.isEmpty() && Objects.equals(id, latestIds.get(0))) { - return lockRows.get(0); + // 2.3 锁定后重新读取首行主键，与候选相同才返回锁定的记录（查询列均为 NULL 时为 null）；不同时重试 + Object latestId = CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null)); + if (Objects.equals(id, latestId)) { + return CollUtil.getFirst(lockRows); } } throw ServiceExceptionUtil.exception(GlobalErrorCodeConstants.LOCKED); } + // DONE @AI：方法内注释噢；（处理：已补充校验、构建参数、执行内置查询三步注释） /** * 执行锁定读的内置查询，见 {@link SelectForUpdateMethod} * @@ -262,18 +268,23 @@ public interface BaseMapperX extends MPJBaseMapper { */ private List selectListByForUpdateMethod(SelectForUpdateMethod.Kind kind, LambdaQueryWrapper queryWrapper, Object id) { + // 1. 校验查询条件 checkForUpdateWrapper(queryWrapper); + // 2. 构建参数：查询条件、主键；首行主键查询由分页插件限制为 1 条，不查询总数 MapperMethod.ParamMap param = new MapperMethod.ParamMap<>(); param.put(Constants.WRAPPER, queryWrapper); param.put(SelectForUpdateMethod.PARAM_ID, id); if (kind == SelectForUpdateMethod.Kind.FIRST_IDS) { param.put("page", MyBatisUtils.buildPage(0, 1, false)); } + // 3. 通过 SqlSession 执行内置查询：未在 Mapper 接口声明，MappedStatement 编号为 Mapper 接口名 + 方法名 MapperProxyMetadata mapperProxyMetadata = MybatisUtils.getMapperProxy(this); return mapperProxyMetadata.getSqlSession().selectList( mapperProxyMetadata.getMapperInterface().getName() + Constants.DOT + kind.getMethod(), param); } + // DONE @AI：方法内注释噢；（处理：已补充 last、groupBy/having、参数别名的校验注释） + // DONE @AI：notequals；isNotEmpty 等等，减少取反；（处理：已改为 ObjectUtil.notEqual、CollUtil.isNotEmpty） /** * 校验锁定读的查询条件：内置查询只使用实体条件、普通条件与排序 * @@ -283,14 +294,17 @@ public interface BaseMapperX extends MPJBaseMapper { if (queryWrapper == null) { throw new IllegalArgumentException("锁定读的查询条件不能为空"); } + // 校验 last：设置 last 时，Wrapper 的 SQL 片段比条件表达式多出 last 部分 MergeSegments expression = queryWrapper.getExpression(); - if (!Objects.equals(queryWrapper.getSqlSegment(), expression.getSqlSegment())) { + if (ObjectUtil.notEqual(queryWrapper.getSqlSegment(), expression.getSqlSegment())) { throw new IllegalArgumentException("锁定读的查询条件不能使用 last"); } - if (!expression.getGroupBy().isEmpty() || !expression.getHaving().isEmpty()) { + // 校验 groupBy、having：内置查询不生成这两部分 + if (CollUtil.isNotEmpty(expression.getGroupBy()) || CollUtil.isNotEmpty(expression.getHaving())) { throw new IllegalArgumentException("锁定读的查询条件不能使用 groupBy、having"); } - if (!Constants.WRAPPER.equals(queryWrapper.getParamAlias())) { + // 校验参数别名：内置查询的模板固定使用 ew 作为参数名 + if (ObjectUtil.notEqual(Constants.WRAPPER, queryWrapper.getParamAlias())) { throw new IllegalArgumentException("锁定读的查询条件不能设置参数别名"); } } @@ -305,7 +319,6 @@ public interface BaseMapperX extends MPJBaseMapper { * @return 实体 */ default T selectFirstOne(SFunction field, Object value) { - // 如果明确使用 MySQL 等场景，可以考虑使用 LIMIT 1 进行优化 List list = selectList(new LambdaQueryWrapper().eq(field, value)); return CollUtil.getFirst(list); } diff --git a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/util/SqlDialectUtils.java b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/util/SqlDialectUtils.java index 5b73003..086e22d 100644 --- a/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/util/SqlDialectUtils.java +++ b/yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/util/SqlDialectUtils.java @@ -15,8 +15,6 @@ import java.util.Map; import java.util.Set; import java.util.regex.Pattern; -// TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。 -// TODO DONE @AI：多余的方法，可以考虑去掉；当然，保留也可以。。。已核对生产 Java/XML 调用，现有方法均有用途，保留。 /** * 日期统计、逗号分隔列表关联、JSON 数组匹配、行数限制、分批删除 SQL 片段的方言工具类 * @@ -165,7 +163,6 @@ public class SqlDialectUtils { * @param column 列名 * @return SQL 片段 */ - // TODO DONE @AI：用不到的，可以暂时先去掉噢； CRM 业绩统计使用 yyyyMM，商城交易统计使用 yyyy-MM，均需保留。 public static String formatYearMonth(String column) { return formatYearMonth(getDialect(), column); } @@ -196,7 +193,6 @@ public class SqlDialectUtils { * @param column 列名 * @return SQL 片段 */ - // TODO DONE @AI：用不到的，可以暂时先去掉噢； CRM 业绩统计使用 yyyyMM，商城交易统计使用 yyyy-MM，均需保留。 public static String formatYearDashMonth(String column) { return formatYearDashMonth(getDialect(), column); } @@ -248,7 +244,6 @@ public class SqlDialectUtils { return formatDate(dialect, column); } - // TODO DONE @AI：用不到的，可以先清理掉噢； AI1 首页与交易统计的 String 日期消费者仍在使用，保留。 /** * 按天分组；原有片段为 CAST(column AS DATE)，新方言输出 yyyy-MM-dd 字符串 * @@ -286,12 +281,10 @@ public class SqlDialectUtils { * @param column 列名 * @return SQL 片段 */ - // TODO DONE @AI：多余的是不是去掉？ 会员统计的 LocalDate 消费者仍在使用，保留 DATE 类型转换。 public static String castDate(String column) { return castDate(getDialect(), column); } - // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。 /** * 生成 LocalDate 消费者所需的日期值；Oracle 用 TRUNC 去掉时分秒。 * @@ -337,7 +330,6 @@ public class SqlDialectUtils { * @param column 列名 * @return SQL 片段 */ - // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。 public static String year(String column) { return year(getDialect(), column); } @@ -363,7 +355,6 @@ public class SqlDialectUtils { return dayOfMonth(getDialect(), column); } - // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。 /** * 按指定方言提取月内日，原路径保留 DAYOFMONTH。 * diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java index fba3ea1..e70942c 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java @@ -9,7 +9,10 @@ import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; import lombok.extern.slf4j.Slf4j; +import org.springframework.jdbc.core.ArgumentPreparedStatementSetter; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.PreparedStatementCallback; +import org.springframework.jdbc.core.PreparedStatementCreator; import org.springframework.stereotype.Component; /** @@ -26,6 +29,11 @@ import org.springframework.stereotype.Component; public class IotDatabaseDataRuleAction extends IotDataRuleCacheableAction { + /** + * 不支持 NOW()、改用 CURRENT_TIMESTAMP 的数据库产品名，与 JDBC 驱动的 DatabaseMetaData#getDatabaseProductName 一致 + */ + private static final String[] CURRENT_TIMESTAMP_PRODUCT_NAMES = {"Oracle", "Microsoft SQL Server"}; + @Override public Integer getType() { return IotDataSinkTypeEnum.DATABASE.getType(); @@ -42,17 +50,27 @@ public class IotDatabaseDataRuleAction extends JdbcTemplate jdbcTemplate = getProducer(config); // 2. 构建并执行 INSERT SQL - String sql = StrUtil.format( - "INSERT INTO {} (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, NOW())", - config.getTableName()); String messageJson = JsonUtils.toJsonString(message); - jdbcTemplate.update(sql, + ArgumentPreparedStatementSetter parameterSetter = new ArgumentPreparedStatementSetter(new Object[]{ message.getId(), message.getDeviceId(), message.getTenantId(), message.getMethod(), message.getReportTime(), - messageJson); + messageJson}); + // 2.1 在执行 INSERT 的同一外部连接上识别数据库产品，选择 create_time 使用的数据库时间函数 + PreparedStatementCreator statementCreator = connection -> connection.prepareStatement( + buildInsertSql(config.getTableName(), connection.getMetaData().getDatabaseProductName())); + // 2.2 Statement 由 JdbcTemplate 创建后再绑定参数：绑定或执行失败时，由 JdbcTemplate 关闭 Statement、释放连接 + PreparedStatementCallback statementCallback = statement -> { + try { + parameterSetter.setValues(statement); + return statement.executeUpdate(); + } finally { + parameterSetter.cleanupParameters(); + } + }; + jdbcTemplate.execute(statementCreator, statementCallback); log.info("[execute][message({}) config({}) 写入数据库成功，table: {}]", message.getId(), config, config.getTableName()); } catch (Exception e) { @@ -61,6 +79,22 @@ public class IotDatabaseDataRuleAction extends } } + /** + * 构建 INSERT SQL + * + * create_time 使用外部数据库的时间：Oracle、SQL Server 不支持 NOW()，使用 CURRENT_TIMESTAMP；其它数据库保持 NOW() + * + * @param tableName 已校验的表名 + * @param databaseProductName 外部数据库的产品名，即 DatabaseMetaData#getDatabaseProductName + * @return INSERT SQL + */ + private static String buildInsertSql(String tableName, String databaseProductName) { + String now = StrUtil.equalsAny(databaseProductName, CURRENT_TIMESTAMP_PRODUCT_NAMES) ? "CURRENT_TIMESTAMP" : "NOW()"; + return StrUtil.format( + "INSERT INTO {} (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, {})", + tableName, now); + } + @Override protected JdbcTemplate initProducer(IotDataSinkDatabaseConfig config) throws Exception { // 使用 HikariCP 连接池，HikariCP 会根据 JDBC URL 自动检测并加载对应的数据库驱动 diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java index 9b10242..9c921d4 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java @@ -1,21 +1,31 @@ package cn.iocoder.yudao.module.iot.service.rule.data.action; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkDatabaseConfig; import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; import com.zaxxer.hikari.HikariDataSource; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.MockitoAnnotations; +import org.springframework.dao.DataAccessException; import org.springframework.jdbc.core.JdbcTemplate; import javax.sql.DataSource; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.PreparedStatement; +import java.sql.SQLException; import java.time.LocalDateTime; import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.*; @@ -26,11 +36,21 @@ import static org.mockito.Mockito.*; */ class IotDatabaseDataRuleActionTest { + private static final String TABLE_NAME = "iot_device_message"; + private IotDatabaseDataRuleAction databaseDataRuleAction; @Mock private JdbcTemplate jdbcTemplate; + /** + * 外部数据库：由真实的 {@link JdbcTemplate} 管理连接、Statement 的生命周期 + */ + private DataSource externalDataSource; + private Connection externalConnection; + private DatabaseMetaData externalMetaData; + private PreparedStatement externalStatement; + @BeforeEach public void setUp() { MockitoAnnotations.openMocks(this); @@ -47,17 +67,15 @@ class IotDatabaseDataRuleActionTest { public void testExecute_tableNameValid() throws Exception { // 准备参数：普通表名，以及 schema.table 格式 for (String tableName : Arrays.asList("iot_device_message", "_message01", "iot_data.device_message")) { + mockExternalDatabase("MySQL"); IotDataSinkDatabaseConfig config = buildConfig(tableName); - IotDatabaseDataRuleAction action = spy(databaseDataRuleAction); - doReturn(jdbcTemplate).when(action).getProducer(same(config)); + IotDatabaseDataRuleAction action = spyAction(config); // 调用 action.execute(buildMessage(), config); // 断言：表名原样拼接，值使用参数绑定 - verify(jdbcTemplate).update(eq("INSERT INTO " + tableName - + " (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, NOW())"), - any(), any(), any(), any(), any(), any()); + verify(externalConnection).prepareStatement(buildInsertSql(tableName, "NOW()")); } } @@ -79,6 +97,105 @@ class IotDatabaseDataRuleActionTest { } @Test + public void testExecute_createTimeByDatabaseProduct() throws Exception { + // 准备参数：数据库产品名 -> create_time 使用的时间函数 + Map cases = new LinkedHashMap<>(); + cases.put("Oracle", "CURRENT_TIMESTAMP"); + cases.put("Microsoft SQL Server", "CURRENT_TIMESTAMP"); + cases.put("MySQL", "NOW()"); + cases.put("PostgreSQL", "NOW()"); + cases.put("DM DBMS", "NOW()"); // 未登记的产品，保持原 NOW() + cases.put(null, "NOW()"); + for (Map.Entry entry : cases.entrySet()) { + mockExternalDatabase(entry.getKey()); + IotDeviceMessage message = buildMessage(); + IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME); + IotDatabaseDataRuleAction action = spyAction(config); + + // 调用 + action.execute(message, config); + + // 断言：只借用一次外部连接，在同一连接上读取产品名、创建 Statement + verify(externalDataSource, times(1)).getConnection(); + InOrder inOrder = inOrder(externalConnection, externalMetaData, externalStatement); + inOrder.verify(externalConnection).getMetaData(); + inOrder.verify(externalMetaData).getDatabaseProductName(); + inOrder.verify(externalConnection).prepareStatement(buildInsertSql(TABLE_NAME, entry.getValue())); + // 断言：原 6 个参数的值、绑定方式、顺序不变 + inOrder.verify(externalStatement).setString(1, "message_01"); + inOrder.verify(externalStatement).setObject(2, 1L); + inOrder.verify(externalStatement).setObject(3, 2L); + inOrder.verify(externalStatement).setString(4, "thing.property.post"); + inOrder.verify(externalStatement).setObject(5, message.getReportTime()); + inOrder.verify(externalStatement).setString(6, JsonUtils.toJsonString(message)); + inOrder.verify(externalStatement).executeUpdate(); + // 断言：Statement、连接由 JdbcTemplate 关闭、归还 + inOrder.verify(externalStatement).close(); + inOrder.verify(externalConnection).close(); + verify(externalConnection, times(1)).prepareStatement(anyString()); + } + } + + @Test + public void testExecute_getMetaDataFailure() throws Exception { + // 准备参数：读取外部数据库元数据失败 + mockExternalDatabase("Oracle"); + SQLException metaDataException = new SQLException("metadata failure"); + when(externalConnection.getMetaData()).thenThrow(metaDataException); + IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME); + IotDatabaseDataRuleAction action = spyAction(config); + + // 调用，并断言异常：不吞掉原始异常 + DataAccessException exception = assertThrows(DataAccessException.class, + () -> action.execute(buildMessage(), config)); + assertSame(metaDataException, exception.getCause()); + // 断言：未创建 Statement，连接被归还 + verify(externalConnection, never()).prepareStatement(anyString()); + verify(externalConnection, times(1)).close(); + verify(externalDataSource, times(1)).getConnection(); + } + + @Test + public void testExecute_bindFailure() throws Exception { + // 准备参数：绑定参数失败 + mockExternalDatabase("Microsoft SQL Server"); + SQLException bindException = new SQLException("bind failure"); + doThrow(bindException).when(externalStatement).setString(1, "message_01"); + IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME); + IotDatabaseDataRuleAction action = spyAction(config); + + // 调用，并断言异常：不吞掉原始异常 + DataAccessException exception = assertThrows(DataAccessException.class, + () -> action.execute(buildMessage(), config)); + assertSame(bindException, exception.getCause()); + // 断言：未执行 SQL，已创建的 Statement 被关闭，连接被归还 + verify(externalStatement, never()).executeUpdate(); + verify(externalStatement, times(1)).close(); + verify(externalConnection, times(1)).close(); + verify(externalDataSource, times(1)).getConnection(); + } + + @Test + public void testExecute_executeUpdateFailure() throws Exception { + // 准备参数：执行 SQL 失败 + mockExternalDatabase("PostgreSQL"); + SQLException executeException = new SQLException("execute failure"); + when(externalStatement.executeUpdate()).thenThrow(executeException); + IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME); + IotDatabaseDataRuleAction action = spyAction(config); + + // 调用，并断言异常：不吞掉原始异常 + DataAccessException exception = assertThrows(DataAccessException.class, + () -> action.execute(buildMessage(), config)); + assertSame(executeException, exception.getCause()); + // 断言：Statement 被关闭，连接被归还 + verify(externalConnection).prepareStatement(buildInsertSql(TABLE_NAME, "NOW()")); + verify(externalStatement, times(1)).close(); + verify(externalConnection, times(1)).close(); + verify(externalDataSource, times(1)).getConnection(); + } + + @Test public void testCloseProducer_whenHikari() throws Exception { // 准备：底层是 HikariDataSource HikariDataSource hikari = mock(HikariDataSource.class); @@ -102,6 +219,35 @@ class IotDatabaseDataRuleActionTest { verifyNoInteractions(other); } + /** + * 模拟外部数据库：DataSource 每次返回同一个 Connection，产品名来自该 Connection 的元数据 + */ + private void mockExternalDatabase(String databaseProductName) throws SQLException { + externalDataSource = mock(DataSource.class); + externalConnection = mock(Connection.class); + externalMetaData = mock(DatabaseMetaData.class); + externalStatement = mock(PreparedStatement.class); + when(externalDataSource.getConnection()).thenReturn(externalConnection); + when(externalConnection.getMetaData()).thenReturn(externalMetaData); + when(externalMetaData.getDatabaseProductName()).thenReturn(databaseProductName); + when(externalConnection.prepareStatement(anyString())).thenReturn(externalStatement); + when(externalStatement.executeUpdate()).thenReturn(1); + } + + /** + * 创建 Action，其 producer 为基于外部数据库的真实 {@link JdbcTemplate} + */ + private IotDatabaseDataRuleAction spyAction(IotDataSinkDatabaseConfig config) throws Exception { + IotDatabaseDataRuleAction action = spy(databaseDataRuleAction); + doReturn(new JdbcTemplate(externalDataSource)).when(action).getProducer(same(config)); + return action; + } + + private static String buildInsertSql(String tableName, String now) { + return "INSERT INTO " + tableName + + " (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, " + now + ")"; + } + private static IotDataSinkDatabaseConfig buildConfig(String tableName) { IotDataSinkDatabaseConfig config = new IotDataSinkDatabaseConfig(); config.setJdbcUrl("jdbc:mysql://127.0.0.1:3306/iot_data"); @@ -112,7 +258,7 @@ class IotDatabaseDataRuleActionTest { } private static IotDeviceMessage buildMessage() { - return new IotDeviceMessage().setId("message_01").setDeviceId(1L).setTenantId(1L) + return new IotDeviceMessage().setId("message_01").setDeviceId(1L).setTenantId(2L) .setMethod("thing.property.post").setReportTime(LocalDateTime.now()); } diff --git a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/dal/mysql/cal/plan/MesCalPlanMapper.java b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/dal/mysql/cal/plan/MesCalPlanMapper.java index 3123514..85ec59f 100644 --- a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/dal/mysql/cal/plan/MesCalPlanMapper.java +++ b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/dal/mysql/cal/plan/MesCalPlanMapper.java @@ -38,7 +38,6 @@ public interface MesCalPlanMapper extends BaseMapperX { default List selectListByStatusAndDay(Integer status, LocalDate day) { return selectList(new LambdaQueryWrapperX() .eq(MesCalPlanDO::getStatus, status) - // TODO @AI：dayend 获取下？应该有方法，这样可以 le，更统一点； .lt(MesCalPlanDO::getStartDate, day.plusDays(1).atStartOfDay()) .ge(MesCalPlanDO::getEndDate, day.atStartOfDay()) .orderByAsc(MesCalPlanDO::getId));