Java: MultiDataSource 依赖 apijson-milvus 来简化代码

TommyLemon · TommyLemon · commit 99dadd9696da · 2024-01-06T20:31:17.000+08:00
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/libs/apijson-milvus-1.0.0.jar b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/libs/apijson-milvus-1.0.0.jar
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/pom.xml b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/pom.xml
@@ -73,6 +73,11 @@
             <artifactId>apijson-router</artifactId>
             <version>1.8.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.github.APIJSON</groupId>
+            <artifactId>apijson-milvus</artifactId>
+            <version>1.0.0</version>
+        </dependency>
         <dependency>
             <groupId>com.github.APIJSON</groupId>
             <artifactId>apijson-mongodb</artifactId>
@@ -148,10 +153,13 @@
             <groupId>org.datayoo.moql</groupId>
             <artifactId>moql-querier-milvus</artifactId>
             <version>1.0.0</version>
+            <groupId>io.milvus</groupId>
+            <artifactId>milvus-sdk-java</artifactId>
+            <version>2.2.2</version>
             <exclusions>
                 <exclusion>
-                        <groupId>org.apache.logging.log4j</groupId>
-                        <artifactId>log4j-slf4j-impl</artifactId>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-slf4j-impl</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
@@ -194,6 +202,24 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.datayoo.moql</groupId>
+            <artifactId>moql-meta</artifactId>
+            <version>1.1.8</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.datayoo.moql</groupId>
+            <artifactId>moql-engine</artifactId>
+            <version>1.4.6</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.datayoo.moql</groupId>
+            <artifactId>moql-querier</artifactId>
+            <version>1.2.3</version>
+            <scope>compile</scope>
+        </dependency>
 
     </dependencies>
 
@@ -237,14 +263,14 @@
 
         <repository>
             <id>spring-snapshots</id>
-            <url>http://repo.spring.io/snapshot</url>
+            <url>https://repo.spring.io/snapshot</url>
             <snapshots>
                 <enabled>true</enabled>
             </snapshots>
         </repository>
         <repository>
             <id>spring-milestones</id>
-            <url>http://repo.spring.io/milestone</url>
+            <url>https://repo.spring.io/milestone</url>
             <snapshots>
                 <enabled>true</enabled>
             </snapshots>
@@ -253,11 +279,11 @@
     <pluginRepositories>
         <pluginRepository>
             <id>spring-snapshots</id>
-            <url>http://repo.spring.io/snapshot</url>
+            <url>https://repo.spring.io/snapshot</url>
         </pluginRepository>
         <pluginRepository>
             <id>spring-milestones</id>
-            <url>http://repo.spring.io/milestone</url>
+            <url>https://repo.spring.io/milestone</url>
         </pluginRepository>
     </pluginRepositories>
 
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLConfig.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLConfig.java
@@ -23,7 +23,9 @@
 import java.util.*;
 
 import apijson.*;
+import apijson.orm.AbstractParser;
 import apijson.orm.AbstractSQLConfig;
+import apijson.orm.Parser;
 import com.alibaba.fastjson.annotation.JSONField;
 
 import apijson.column.ColumnUtil;
@@ -72,11 +74,17 @@ public String getQuote() {
 
 	@Override
 	public String getLimitString() {
-		if (DATABASE_MILVUS.equals(getDatabase()) && RequestMethod.isGetMethod(getMethod(), true)) {
+		if (isMilvus()) {
 			int count = getCount();
+			if (count == 0) {
+				Parser<Long> parser = getParser();
+				count = parser == null ? AbstractParser.MAX_QUERY_COUNT : parser.getMaxQueryCount();
+			}
+
 			int offset = getOffset(getPage(), count);
-			return " LIMIT " + offset + ", " + count;
+			return " LIMIT " + offset + ", " + count; // 目前 moql-transx 的限制
 		}
+
 		return super.getLimitString();
 	}
 
@@ -210,7 +218,8 @@ public String getDBUri() {
 		}
 
 		if (isMySQL()) {
-			// 这个是 MySQL 8.0 及以上，要加 userSSL=false  return "jdbc:mysql://localhost:3306?userSSL=false&serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8";
+			// 这个是 MySQL 8.0 及以上，要加 userSSL=false
+//			 return "jdbc:mysql://47.122.25.116:3306?userSSL=false&serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8";
 			// 以下是 MySQL 5.7 及以下
 			return "jdbc:mysql://localhost:3306?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8"; //TODO 改成你自己的，TiDB 可以当成 MySQL 使用，默认端口为 4000
 		}
@@ -246,27 +255,9 @@ public String getDBUri() {
 			return "jdbc:mongodb://atlas-sql-6593c65c296c5865121e6ebe-xxskv.a.query.mongodb.net/myVirtualDatabase?ssl=true&authSource=admin";
 		}
 
-		return "";
-	}
-
-
-	// TODO 迁移到 APIJSON 主项目 <<<<<<<<<<<<<<<<<<<<
-	@Override
-	public String getSchema() {
-		String sch = super.getSchema();
-		if (StringUtil.isEmpty(sch) && isInfluxDB()) {
-			sch = DEFAULT_SCHEMA;
-		}
-		return sch;
-	}
-
-	@Override
-	public String getSQLSchema() {
-		return isInfluxDB() ? null : super.getSQLSchema();
+		return null;
 	}
 
-	// TODO 迁移到 APIJSON 主项目 >>>>>>>>>>>>>>>>>>>>>>
-
 	private String dbAccount;
 	public DemoSQLConfig setDBAccount(String dbAccount) {
 		this.dbAccount = dbAccount;
@@ -457,6 +448,24 @@ protected void onJoinComplexRelation(String sql, String quote, Join join, String
 	}
 
 
+
+	// TODO 迁移到 apijson-influxdb 主项目 <<<<<<<<<<<<<<<<<<<<
+	@Override
+	public String getSchema() {
+		String sch = super.getSchema();
+		if (StringUtil.isEmpty(sch) && isInfluxDB()) {
+			sch = DEFAULT_SCHEMA;
+		}
+		return sch;
+	}
+
+	@Override
+	public String getSQLSchema() {
+		return isInfluxDB() ? null : super.getSQLSchema();
+	}
+
+	// TODO 迁移到 apijson-influxdb 主项目 >>>>>>>>>>>>>>>>>>>>>>
+
 	@Override
 	public String getSQLTable() {
 		String t = super.getSQLTable();
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLExecutor.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLExecutor.java
@@ -15,6 +15,7 @@
 package apijson.demo;
 
 import apijson.*;
+import apijson.milvus.MilvusUtil;
 import apijson.mongodb.MongoUtil;
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.fastjson.JSONObject;
@@ -43,6 +44,7 @@
 import org.datayoo.moql.ColumnDefinition;
 import org.datayoo.moql.RecordSet;
 import org.datayoo.moql.RecordSetDefinition;
+import org.datayoo.moql.querier.DataQuerier;
 import org.datayoo.moql.querier.milvus.MilvusQuerier;
 import org.influxdb.BatchOptions;
 import org.influxdb.InfluxDB;
@@ -205,70 +207,12 @@ public Connection getConnection(SQLConfig<Long> config) throws Exception {
 
     @Override
     public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType) throws Exception {
-        if (DATABASE_MILVUS.equals(config.getDatabase())) {  // 3.0.0 及以下要这样连接
-            String uri = config.getDBUri();
-//
-//            int start = uri.indexOf("://");
-//            String prefix = uri.substring(0, start);
-//
-//            uri = uri.substring(start + "://".length());
-//            int end = uri.indexOf(":");
-//            int port = Integer.parseInt(uri.substring(end + 1));
-//            String host = uri.substring(0, end);
-
-            // 构建Milvus客户端
-            MilvusServiceClient milvusClient = new MilvusServiceClient(
-                    ConnectParam.newBuilder().withUri(uri).build()
-            );
-
-            // 使用Milvus客户端创建Milvus查询器
-            MilvusQuerier milvusQuerier = new MilvusQuerier(milvusClient);
-
-            /*
-            查询语句含义：从book集合中筛选数据，并返回col1,col2两个列。筛选条件为，当数据的col3列值为4，col4列值为'a','b','c'中的任意一
-            个，且vec向量字段采用'L2'类型匹配，值为'[[1.0, 2.0, 3.0],[1.1,2.1,3.1]]'。另外，采用强一致性级别在10个单元内进行检索，取第11到第15，5条命中记录。
-            */
-            String sql = config.getSQL(false); //
-//            String sql = "select id,userId,momentId,content,date from Comment where vMatch(vec, 'L2', '[[1]]') and consistencyLevel('STRONG')  limit 1,1";
-            // 使用查询器执行sql语句，并返回查询结果
-            RecordSet recordSet = milvusQuerier.query(sql);
-
-//            int count = recordSet == null ? 0 : recordSet.getRecordsCount();
-            List<Map<String, Object>> list = recordSet == null ? null : recordSet.getRecordsAsMaps();
-//            RecordSetDefinition def = recordSet.getRecordSetDefinition();
-//            List<ColumnDefinition> cols = def.getColumns();
-
-//            List<Object[]> list = count <= 0 ? null : recordSet.getRecords();
-
-            if (list == null || list.isEmpty()) {
-                return new JSONObject(true);
-            }
-
-            List<JSONObject> nl = new ArrayList<>(list.size());
-            for (int i = 0; i < list.size(); i++) {
-                Map<String, Object> map = list.get(i);
-
-                JSONObject obj = new JSONObject(map == null ? new HashMap<>() : map);
-                // obj.put(col.getValue(), os[j]);
-//                for (int j = 0; j < os.length; j++) {
-//                    ColumnDefinition col = cols.get(j);
-//                    obj.put(col.getValue(), os[j]);
-//                }
-                nl.add(obj);
-            }
-
-            JSONObject result = nl.get(0); // JSON.parseObject(list.get(0));
-            if (nl.size() > 1) {
-                result.put(KEY_RAW_LIST, nl);
-            }
-
-            return result;
-        }
-
+        boolean isMilvus = DATABASE_MILVUS.equals(config.getDatabase()); // APIJSON 6.4.0+ 可用 config.isMilvus();
         boolean isCassandra = config.isCassandra();
         boolean isInfluxDB = config.isInfluxDB();
 
-        if (isCassandra || isInfluxDB) {
+        if (isMilvus || isCassandra || isInfluxDB) {
+            // TODO 把 execute 内与缓存无关只与数据库读写逻辑相关的代码抽取到 executeSQL 函数
             String sql = config.getSQL(false); // config.isPrepared());
             List<JSONObject> cache = getCache(sql, config);
             int position = config.getPosition();
@@ -288,13 +232,18 @@ public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType)
                 isWrite = sqlPrefix.startsWith("INSERT ") || sqlPrefix.startsWith("UPDATE ") || sqlPrefix.startsWith("DELETE ");
             }
 
+            List<JSONObject> resultList = new ArrayList<>();
+
+            if (isMilvus) {
+                return MilvusUtil.execute(config, unknownType);
+            }
 
             if (isCassandra) {
                 CqlSession session = CqlSession.builder()
 //                        .withCloudSecureConnectBundle(Paths.get("/path/to/secure-connect-database_name.zip"))
                         .withCloudSecureConnectBundle(new URL(config.getDBUri()))
                         .withAuthCredentials(config.getDBAccount(), config.getDBPassword())
-                        .withKeyspace(config.getSchema())
+                        .withKeyspace(config.getSQLSchema())
                         .build();
 
                 //            if (config.isPrepared()) {
@@ -316,16 +265,11 @@ public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType)
                     return new JSONObject(true);
                 }
 
-                result = JSON.parseObject(list.get(0));
-                if (list.size() > 1) {
-                    result.put(KEY_RAW_LIST, list);
+                for (int i = 0; i < list.size(); i++) {
+                    resultList.add(JSON.parseObject(list.get(i)));
                 }
-
-                return result;
             }
-
-
-            if (isInfluxDB) {
+            else if (isInfluxDB) {
                 InfluxDB influxDB = InfluxDBFactory.connect(config.getDBUri(), config.getDBAccount(), config.getDBPassword());
                 influxDB.setDatabase(config.getSchema());
 
@@ -382,8 +326,6 @@ public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType)
                     return new JSONObject(true);
                 }
 
-                List<JSONObject> resultList = new ArrayList<>();
-
                 for (int i = 0; i < list.size(); i++) {
                     QueryResult.Result qyrt = list.get(i);
                     List<QueryResult.Series> seriesList = qyrt.getSeries();
@@ -413,16 +355,17 @@ public JSONObject execute(@NotNull SQLConfig<Long> config, boolean unknownType)
                     }
                 }
 
-                result = resultList.isEmpty() ? new JSONObject() : resultList.get(0);
-                if (resultList.size() > 1) {
-                    result.put(KEY_RAW_LIST, resultList);
-                }
-                
-                putCache(sql, resultList, config);
+            }
 
-                return result;
+            // TODO 把 execute 内与缓存无关只与数据库读写逻辑相关的代码抽取到 executeSQL 函数
+            result = resultList.isEmpty() ? new JSONObject() : resultList.get(0);
+            if (resultList.size() > 1) {
+                result.put(KEY_RAW_LIST, resultList);
             }
 
+            putCache(sql, resultList, config);
+
+            return result;
         }
 
         return super.execute(config, unknownType);
