Skip to content

Commit edb37d7

Browse files
TommyLemonTommyLemon
committed
Java：MultiDataSource 通过插件 apijson-iotdb 来简化物联网数据库 Apache/IoTDB 的使用
1 parent 5f438f2 commit edb37d7

File tree

2 files changed

+19
-133
lines changed

2 files changed

+19
-133
lines changed

‎APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLConfig.java‎

Lines changed: 14 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424

2525
import apijson.*;
2626
//import apijson.influxdb.InfluxDBUtil;
27+
import apijson.iotdb.IoTDBUtil;
2728
import apijson.orm.AbstractParser;
2829
import apijson.orm.AbstractSQLConfig;
2930
import apijson.orm.Parser;
@@ -51,69 +52,7 @@ public DemoSQLConfig(RequestMethod method, String table) {
5152
super(method, table);
5253
}
5354

54-
// 支持 NoSQL 数据库 MongoDB，APIJSON 6.4.0- 版本需要手动添加相关代码 <<<<<<<<<<<<<<<<<<<<<<<<<<<<<
55-
public static final String DATABASE_MONGODB = "MONGODB";
56-
public static final String DATABASE_MILVUS = "MILVUS";
57-
public static final String DATABASE_IOTDB = "IOTDB";
58-
59-
@Override
60-
public boolean isPrepared() {
61-
return super.isPrepared() && ! isMongoDB(); // MongoDB JDBC 还不支持预编译
62-
}
63-
64-
public boolean isMongoDB() {
65-
return DATABASE_MONGODB.equals(getDatabase());
66-
67-
}
68-
public boolean isMilvus() {
69-
return DATABASE_MILVUS.equals(getDatabase());
70-
}
71-
public boolean isIoTDB() {
72-
return DATABASE_IOTDB.equals(getDatabase());
73-
}
74-
75-
// MongoDB 同时支持 `tbl` 反引号 和 "col" 双引号
76-
@Override
77-
public String getQuote() {
78-
return isMilvus() ? "`" : (isIoTDB() ? "" : super.getQuote());
79-
}
80-
81-
@Override
82-
public String getLimitString() {
83-
if (isMilvus()) {
84-
int count = getCount();
85-
if (count == 0) {
86-
Parser<Long> parser = getParser();
87-
count = parser == null ? AbstractParser.MAX_QUERY_COUNT : parser.getMaxQueryCount();
88-
}
89-
90-
int offset = getOffset(getPage(), count);
91-
return " LIMIT " + offset + ", " + count; // 目前 moql-transx 的限制
92-
}
93-
94-
return super.getLimitString();
95-
}
96-
9755
static {
98-
DATABASE_LIST.add(DATABASE_MONGODB);
99-
DATABASE_LIST.add(DATABASE_MILVUS);
100-
DATABASE_LIST.add(DATABASE_CASSANDRA);
101-
DATABASE_LIST.add(DATABASE_IOTDB);
102-
103-
// Milvus 需要
104-
SQL_FUNCTION_MAP.put("vMatch", "");
105-
SQL_FUNCTION_MAP.put("consistencyLevel", "");
106-
SQL_FUNCTION_MAP.put("partitionBy", "");
107-
SQL_FUNCTION_MAP.put("gracefulTime", "");
108-
SQL_FUNCTION_MAP.put("guaranteeTimestamp", "");
109-
SQL_FUNCTION_MAP.put("roundDecimal", "");
110-
SQL_FUNCTION_MAP.put("travelTimestamp", "");
111-
SQL_FUNCTION_MAP.put("nProbe", "");
112-
SQL_FUNCTION_MAP.put("ef", "");
113-
SQL_FUNCTION_MAP.put("searchK", "");
114-
115-
// 支持 NoSQL 数据库 MongoDB，APIJSON 6.4.0- 版本需要手动添加相关代码 >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
116-
11756
DEFAULT_DATABASE = DATABASE_MYSQL; //TODO 默认数据库类型，改成你自己的。TiDB, MariaDB, OceanBase 这类兼容 MySQL 的可当做 MySQL 使用
11857
DEFAULT_SCHEMA = "sys"; // ""apijson"; //TODO 默认数据库名/模式，改成你自己的，默认情况是 MySQL: sys, PostgreSQL: sys, SQL Server: dbo, Oracle:
11958

@@ -292,7 +231,7 @@ public String getDBUri() {
292231
return "http://localhost:19530";
293232
}
294233
if (isIoTDB()) {
295-
return "jdbc:iotdb://127.0.0.1:6667?charset=GB18030";
234+
return "jdbc:iotdb://localhost:6667"; // ?charset=GB18030 加参数会报错 URI 格式错误
296235
}
297236
if (isMongoDB()) {
298237
return "jdbc:mongodb://atlas-sql-6593c65c296c5865121e6ebe-xxskv.a.query.mongodb.net/myVirtualDatabase?ssl=true&authSource=admin";
@@ -468,6 +407,8 @@ public void setSql(String sql) {
468407
// public boolean isDb2() {
469408
// return false;
470409
// }
410+
//
411+
// 其它数据库 ...
471412

472413

473414
// 取消注释来兼容 Oracle DATETIME, TIMESTAMP 等日期时间类型的值来写库。5.0.0+ 重写以下方法，4.9.1 及以下改为重写 getValue(String)
@@ -514,15 +455,15 @@ protected int getMaxCombineCount() {
514455

515456

516457
// TODO 迁移到 apijson-influxdb 主项目 <<<<<<<<<<<<<<<<<<<<
517-
// @Override
518-
// public String getSchema() {
519-
// return InfluxDBUtil.getSchema(super.getSchema(), DEFAULT_SCHEMA, isInfluxDB());
520-
// }
521-
//
522-
// @Override
523-
// public String getSQLSchema() {
524-
// return InfluxDBUtil.getSQLSchema(super.getSQLSchema(), isInfluxDB());
525-
// }
458+
@Override
459+
public String getSchema() {
460+
return IoTDBUtil.getSchema(super.getSchema(), DEFAULT_SCHEMA, isIoTDB());
461+
}
462+
463+
@Override
464+
public String getSQLSchema() {
465+
return IoTDBUtil.getSQLSchema(super.getSQLSchema().replaceAll("-", "."), isIoTDB());
466+
}
526467

527468
// TODO 迁移到 apijson-influxdb 主项目 >>>>>>>>>>>>>>>>>>>>>>
528469

@@ -535,8 +476,7 @@ public String getSQLTable() {
535476

536477
@Override
537478
public String getTablePath() {
538-
String p = super.getTablePath();
539-
return isIoTDB() ? p + ".**" : p;
479+
return IoTDBUtil.getTablePath(super.getTablePath(), isIoTDB());
540480
}
541481

542482
// 取消注释可将前端传参驼峰命名转为蛇形命名 aBCdEfg => upper ? A_B_CD_EFG : a_b_cd_efg

‎APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLExecutor.java‎

Lines changed: 5 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,10 @@
2121
//import apijson.influxdb.InfluxDBUtil;
2222
//import apijson.milvus.MilvusUtil;
2323
//import apijson.mongodb.MongoUtil;
24+
import apijson.iotdb.IoTDBUtil;
2425
import apijson.orm.SQLConfig;
2526
import com.alibaba.druid.pool.DruidDataSource;
2627
import com.alibaba.fastjson.JSONObject;
27-
import org.apache.iotdb.isession.SessionDataSet;
28-
import org.apache.iotdb.session.Session;
29-
import org.apache.iotdb.tsfile.read.common.Field;
30-
import org.apache.iotdb.tsfile.read.common.RowRecord;
3128
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
3229
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
3330
import org.springframework.data.redis.core.RedisTemplate;
@@ -195,10 +192,10 @@ public Connection getConnection(SQLConfig<Long> config) throws Exception {
195192

196193
@Override
197194
public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType) throws Exception {
198-
boolean isMilvus = DATABASE_MILVUS.equals(config.getDatabase()); // APIJSON 6.4.0+ 可用 config.isMilvus();
195+
boolean isMilvus = config.isMilvus(); // DATABASE_MILVUS.equals(config.getDatabase()); // APIJSON 6.4.0+ 可用 config.isMilvus();
199196
boolean isCassandra = config.isCassandra();
200197
boolean isInfluxDB = config.isInfluxDB();
201-
boolean isIoTDB = DemoSQLConfig.DATABASE_IOTDB.equals(config.getDatabase());
198+
boolean isIoTDB = config.isIoTDB(); // DemoSQLConfig.DATABASE_IOTDB.equals(config.getDatabase());
202199

203200
if (isMilvus || isCassandra || isInfluxDB || isIoTDB) {
204201
// TODO 把 execute 内与缓存无关只与数据库读写逻辑相关的代码抽取到 executeSQL 函数
@@ -255,62 +252,11 @@ else if (sqlPrefix.startsWith("DELETE ")) {
255252
// }
256253
// else
257254
if (isIoTDB) {
258-
Session session = new Session.Builder()
259-
.username(config.getDBAccount())
260-
.password(config.getDBPassword())
261-
.build();
262-
session.open();
263-
264255
if (isWrite) {
265-
session.executeNonQueryStatement(sql);
266-
267-
Object id = config.getId();
268-
Object idIn = id != null ? null : config.getIdIn();
269-
Collection<?> ids = idIn instanceof Collection<?> ? (Collection<?>) idIn : null;
270-
int count = id != null ? 1 : (ids == null || ids.isEmpty() ? 1 : ids.size());
271-
272-
result = DemoParser.newSuccessResult();
273-
result.put(JSONResponse.KEY_COUNT, 1);
274-
result.put(JSONResponse.KEY_OK, true);
275-
276-
session.close();
277-
278-
return result;
279-
}
280-
281-
SessionDataSet ds = session.executeQueryStatement(sql);
282-
List<String> ns = ds == null ? null : ds.getColumnNames();
283-
List<String> nameList = ns == null || ns.isEmpty() ? null : new ArrayList<>(ns.size());
284-
if (nameList != null) {
285-
String prefix = config.getSQLSchema() + "." + config.getSQLTable() + ".";
286-
287-
for (String name : ns) {
288-
if (name.startsWith(prefix)) {
289-
name = name.substring(prefix.length());
290-
}
291-
292-
nameList.add(name);
293-
}
294-
295-
resultList = new ArrayList<>(ds.getFetchSize());
296-
297-
while (ds.hasNext()) {
298-
RowRecord row = ds.next();
299-
List<Field> fs = row.getFields();
300-
301-
JSONObject obj = new JSONObject(true);
302-
obj.put(nameList.get(0), row.getTimestamp());
303-
for (int i = 0; i < fs.size(); i++) {
304-
Field f = fs.get(i);
305-
Object v = f == null ? null : f.getObjectValue(f.getDataType());
306-
obj.put(nameList.get(i + 1), v);
307-
}
308-
309-
resultList.add(obj);
310-
}
256+
return IoTDBUtil.executeUpdate(config, sql);
311257
}
312258

313-
session.close();
259+
resultList = IoTDBUtil.executeQuery(config, sql, unknownType);
314260
}
315261

316262
// TODO 把 execute 内与缓存无关只与数据库读写逻辑相关的代码抽取到 executeSQL 函数

0 commit comments

Comments
 (0)