添加了一些注释,日志打印和bug修复

pom.xml

@@ -86,6 +86,12 @@
             <artifactId>graal-sdk</artifactId>
             <version>23.0.0</version>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/cn.com.kingbase/kingbase8 -->
+        <dependency>
+            <groupId>cn.com.kingbase</groupId>
+            <artifactId>kingbase8</artifactId>
+            <version>8.6.0</version>
+        </dependency>
         <dependency>
             <groupId>org.graalvm.js</groupId>
             <artifactId>js</artifactId>
@@ -96,11 +102,11 @@
             <artifactId>js-scriptengine</artifactId>
             <version>23.0.0</version>
         </dependency>
-<!--        <dependency>-->
-<!--            <groupId>org.apache.kafka</groupId>-->
-<!--            <artifactId>kafka-clients</artifactId>-->
-<!--            <version>3.9.1</version>-->
-<!--        </dependency>-->
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>3.9.1</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents.client5</groupId>
             <artifactId>httpclient5-fluent</artifactId>
@@ -123,11 +129,11 @@
             <version>9.4.2.1</version>
         </dependency>
         <!-- RocketMQ -->
-<!--        <dependency>-->
-<!--            <groupId>org.apache.rocketmq</groupId>-->
-<!--            <artifactId>rocketmq-client-java</artifactId>-->
-<!--            <version>5.0.8</version>-->
-<!--        </dependency>-->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client-java</artifactId>
+            <version>5.0.8</version>
+        </dependency>
         <!-- ActiveMQ -->
         <dependency>
             <groupId>org.apache.activemq</groupId>

src/main/java/com/bfkj/unidia/Core/DataServices.java

@@ -40,6 +40,8 @@ public class DataServices {
      * @param params 入口参数（必须包含service_code）
      * @return 处理结果
      */
+// params:     {service_code:001，dataContent:{data:{key1:value1,key2:value2,……},conditions:{key1:value1,key2:value2,……}},page:1,pagesize:20,columns:[key1,key2],permission:{key1:value1,key2:value2,……},event:"0"}
+// columns:[key1,key2] aop处理 查询哪些字段  列权限 , permission (行权限) where 条件条件
     public Result<Map<String, Object>> process(Map<String, Object> params) {
         long startTime = System.nanoTime();
         // 1. 验证参数并获取服务信息
@@ -58,6 +60,7 @@ public class DataServices {
         // 2. 初始化执行上下文--整个算法过程所需要的信息及交互信息
         ProcessContext context = new ProcessContext();
         context.setServiceCode(serviceCode);
+        context.addAlgorithmResult("0000",params);
         //不使用UUID，因为没有可用信息，还是需要时间戳+容器编号+服务编号，方便追踪
 
         // 3. 获取并排序算法配置
@@ -99,6 +102,7 @@ public class DataServices {
                 logger.error("前置条件评估失败:{} ", conditionResult.getError());
                 return Result.fail("前置条件评估失败: " + conditionResult.getError());
             }
+            // 前置条件执行的结果
             int conditionValue = conditionResult.getData();
             if (conditionValue == 1) { // 跳过当前算法
                 logger.info("跳过算法: " + config.algorithmCode());
@@ -107,7 +111,7 @@ public class DataServices {
                 logger.info("跳过算法及后续算法: " + config.algorithmCode());
                 break;
             }
-            // 2. 准备输入参数
+            // 2. 准备输入参数  {"dataList":"@0010.returnData"} ,  {"0010":{"xx":消费到的数据集合}} ,
             Map<String, Object> inputData = parseInputTemplate(config.inputDataTemplate(),  context.getAllResults(),config);
             // 3. 执行算法
             Result<Map<String, Object>> executionResult = executeAlgorithm(config, inputData);
@@ -175,7 +179,7 @@ public class DataServices {
      * 方法首先会检查模板是否为空，若为空则返回一个空的 Map
      * 然后，方法会将模板字符串解析为一个 Map 对象，并通过递归解析方法处理可能存在的嵌套结构和占位符
      * 最后，将解析后的对象转换为 Map 形式返回
-     */
+     */                                          //input_data_template
     private Map<String, Object> parseInputTemplate(String template, Map<String, Object> allResults,AlgorithmConfig config) {
         // 检查模板是否为空，若为空则返回一个空的 Map
         if (template == null || template.isEmpty()) {
@@ -184,6 +188,7 @@ public class DataServices {
         // 将模板字符串解析为一个 Map 对象
         Map<String, Object> templateMap = parseJson(template);
         // 递归解析 Map 中的占位符，并将解析后的对象转换为 Map 形式返回
+
         Object recursiveResolve = recursiveResolve(templateMap, allResults,config);
         return DataFormatConverter.ObjectToMap(recursiveResolve);
     }
@@ -192,17 +197,18 @@ public class DataServices {
      * 此方法的目的是深入解析给定值，如果值是 Map 或 List 类型，则递归解析其所有条目
      * 如果值是字符串，则使用 resolveValue 方法解析它
      *
-     * @param value 要解析的原始值，可以是 Map、List 或字符串
+     * @param templateMap 要解析的原始值，可以是 Map、List 或字符串
      * @param allResults 包含所有解析结果的 Map，用于解析值中的引用
      * @return 解析后的值，类型与输入值相同
      * @throws IllegalArgumentException 如果 Map 的键不是字符串
      */
-    private Object recursiveResolve(Object value, Map<String, Object> allResults,AlgorithmConfig config) {
+//    {"dataList":"@0010.returnData"}
+    private Object recursiveResolve(Object templateMap, Map<String, Object> allResults,AlgorithmConfig config) {
         // 当值是 Map 类型时，创建一个新的 Map 来存储解析后的条目
-        if (value instanceof Map<?, ?>) {
+        if (templateMap instanceof Map<?, ?>) {
             Map<String, Object> resolvedMap = new LinkedHashMap<>();
             // 遍历 Map 中的每个条目，并递归解析每个值
-            for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
+            for (Map.Entry<?, ?> entry : ((Map<?, ?>) templateMap).entrySet()) {
                 Object key = entry.getKey();
                 // 确保 Map 的键是字符串，否则抛出异常
                 if (!(key instanceof String)) {
@@ -211,16 +217,17 @@ public class DataServices {
                 resolvedMap.put((String) key, recursiveResolve(entry.getValue(), allResults,config));
             }
             return resolvedMap;
-        } else if (value instanceof List<?>) {
+        } else if (templateMap instanceof List<?>) {
             // 当值是 List 类型时，对列表中的每个项进行递归解析
-            return ((List<?>) value).stream()
+            return ((List<?>) templateMap).stream()
                     .map(item -> recursiveResolve(item, allResults,config))
                     .collect(Collectors.toList());
         } else {
             // 当值是字符串时，使用 resolveValue 方法解析它
-            return resolveValue( value.toString(), allResults, config);
+            return resolveValue( templateMap.toString(), allResults, config);
         }
     }
+
     /**
      * 解析值模板
      * 本方法根据值模板的前缀字符来决定如何解析值：
@@ -251,7 +258,7 @@ public class DataServices {
      * 从算法结果中获取值
      * 本方法通过路径的方式从一个算法结果集中获取特定的值路径以点号分隔，表示算法代码和键的层次结构
      * 例如，路径"algorithm.code.subKey"将尝试从算法代码"algorithm.code"的结果中获取"subKey"的值
-     *
+     * todo 需要完善
      * @param path 表示值路径的字符串，用于定位算法结果中的特定值
      * @param allResults 包含所有算法结果的映射，键是算法代码，值是该算法的结果对象
      * @return 定位到的值对象，如果无法定位到值或者路径格式不正确，则返回null
@@ -544,7 +551,7 @@ public class DataServices {
      * @return 解析后的Map对象，包含JSON数据的键值对
      * @throws IllegalArgumentException 如果输入的JSON字符串为空，则抛出此异常
      */
-    private Map<String, Object> parseJson(String json) {
+    private static Map<String, Object> parseJson(String json) {
         // 检查输入的JSON字符串是否为空，如果为空则抛出异常
         if (json == null || json.trim().isEmpty()) {
             throw new IllegalArgumentException("JSON输入不能为空");

src/main/java/com/bfkj/unidia/Core/ServiceController.java

@@ -78,7 +78,7 @@ public class ServiceController {
      * @param type 服务或容器的类型标识
      * @return 返回一个包含操作结果的Result对象
      */
-    private Result<Integer> controllerService(String containerCode, String serviceCode,  int type) {
+    private Result<Integer>  controllerService(String containerCode, String serviceCode,  int type) {
         try {
             if(serviceCode == null || serviceCode.isEmpty()) {//服务编号为空默认全部服务
                 return controllerAllService(containerCode,  type);

src/main/java/com/bfkj/unidia/DataBaseUtils/ConnectionPoolManager.java

@@ -118,13 +118,6 @@ public class ConnectionPoolManager {
     private Result<HikariDataSource> createDataSource(String dbKey, String dbType,Map<String, String> dbConfig) {
         try {
             String url = dbConfig.get("url");
-            //如果是sqlite则处理路径
-//            if (dbType.equals("sqlite")) {
-//                String path = System.getProperty("user.dir") + File.separator + "db" + File.separator + "unidia206.db";
-//                url = url.replace("unidia206.db", path);
-//                logger.info("sqlite path: {}", url);
-//            }
-
             // 创建HikariCP配置
             HikariConfig config = new HikariConfig();
 

src/main/java/com/bfkj/unidia/DataBaseUtils/DbExecutor.java

@@ -46,7 +46,7 @@ public class DbExecutor {
      * @return 返回一个Result对象，包含操作结果，主要是影响的行数
      */
     public Result<Integer> dbUpdate(Map<String, String> dbConfig, String tableName, Map<String, Object> params) {
-        //参数预处理，包括参数校验
+        //参数预处理，包括参数校验 (会拼接成完整的sql)
         Result<DbParamsService.ProcessedResult> processDataContextResult = dbParamsService.processExecutorParams(params, dbConfig, tableName);
         //如果参数不合法，返回错误信息
         if (!processDataContextResult.isSuccess()) {
@@ -81,8 +81,10 @@ public class DbExecutor {
                                        DbParamsService.ProcessedResult dbParams){
         try  {
             // 检查是否存在新增字段，若存在则创建表结构
-            Map<String, Object> addColumn = dbParams.getAddColumns();
-            if (addColumn.size() > 0) {
+            Map<String, Object> addColumn = dbParams.getAddColumns(); // "c" : value
+
+            if (addColumn.size() > 0) { // 不存在的列数量是否大于0
+                // 创建表 或者创建字段
                 Result<Integer> createTableResult = tableStructureManager.createTable(dbConfig, tableName, addColumn);
                 if (!createTableResult.isSuccess()) {
                     return createTableResult;
@@ -91,9 +93,9 @@ public class DbExecutor {
 
             // 获取插入的列名和对应的值，执行批量插入操作
             String insertColumns = dbParams.getInsertColumns();
-            List<List<Object>> insertParam = dbParams.getInsertValues();
+            List<List<Object>> insertParam = dbParams.getInsertValues(); // 占位符对应的value集合
             int insertCount = 0;
-            if (insertParam.size() > 0) {
+            if (insertParam.size() > 0) { //  新增
                 Result<Integer> batchInsert = jdbcExecutor.batchUpdate(dbConfig, insertColumns, insertParam);
                 if(!batchInsert.isSuccess()){
                     return Result.fail(batchInsert.getError());
@@ -136,7 +138,7 @@ public class DbExecutor {
             for(String queryCondition : dbParams.getQueryConditions().keySet()){
                 // 获取当前查询条件的参数
                 List<List<Object>> queryParam = dbParams.getQueryConditions().get(queryCondition);
-                // 拼接更新SQL语句
+                // 拼接更新SQL语句 , sql拼接完成
                 String updateSql = updateColumns.concat(" where ").concat(queryCondition);
                 // 执行批量更新操作
                 Result<Integer> batchedUpdate = jdbcExecutor.batchUpdate(dbConfig, updateSql, queryParam);

src/main/java/com/bfkj/unidia/DataBaseUtils/DbParamsService.java

@@ -17,6 +17,9 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.sql.Timestamp;
 
+/**
+ * 把前端传入的参数转换为sql语句
+ */
 @Component
 public class DbParamsService {
     private static final Logger logger = LoggerFactory.getLogger(DbParamsService.class);
@@ -44,7 +47,7 @@ public class DbParamsService {
         try {
              //初始化返回参数
             ProcessedResult result = new ProcessedResult();
-            // 处理基础参数（参数检查、event、page、pageSize、permission）返回列权限
+            // 处理基础参数（参数检查、event、page、pageSize、permission）返回列权限 , dataContent目前没处理
             Result<List<String>> processBaseDataResult = processBaseData(params, tableName,result);
             if(!processBaseDataResult.isSuccess()){
                 return Result.fail(processBaseDataResult.getError());
@@ -67,6 +70,8 @@ public class DbParamsService {
             StringBuilder insertColumns = new StringBuilder();//定义一个新增字段字符串--结构中
             StringBuilder updateColumns = new StringBuilder();//定义一个更新字段字符串S--结构中
             buildColumns(modifyColumns,insertColumns,updateColumns,tableName,dbType);
+            // where 之前的全部处理好了
+
             //设置返回结果
             result.setInsertColumns(insertColumns.toString());
             result.setUpdateColumns(updateColumns.toString());
@@ -157,7 +162,6 @@ public class DbParamsService {
                     addColumn, modifyColumns, ParamsList,dbType);
         } else if (dataContent instanceof List<?> dataContentList) {
             for (Object item : dataContentList) {
-                //processDataContent(item, authColumns, tableStructure,addColumn,dbType,modifyColumns,ParamsList);
                 if(item instanceof Map<?,?> dataContentMap) {
                     processSingleDataContent(dataContentMap, authColumns, tableStructure,
                             addColumn, modifyColumns, ParamsList,dbType);
@@ -197,6 +201,7 @@ public class DbParamsService {
         } else if (conditionsObj instanceof List<?> conditionList) {//复杂条件
             processComplexConditions(conditionList, dbType, queryCondition, queryParam);
         }
+        // 到此为止,where 前后的语句都已经拼接完毕
         ParamsList.add(new RowParams(updateValues, queryCondition, queryParam));//添加到行参数
     }
     /**
@@ -227,7 +232,6 @@ public class DbParamsService {
                         if(modifyColumns == null) modifyColumns = new ArrayList<>();
                         modifyColumns.add(keyStr);//新增或更新不重复字段
                     }
-
                     if(tableStructure == null || tableStructure.isEmpty() || !tableStructure.contains(keyStr)) addColumn.put(keyStr, value);//需要创建的字段
                     if(isDateTimeType(value)){//日期时间字段对参数进行数据库适配
                         updateValues.put(keyStr, convertDateTimeValue(value, dbType));
@@ -351,7 +355,7 @@ public class DbParamsService {
         for (RowParams item : ParamsList) {
             String currentEvent = event.toUpperCase();
             if(currentEvent.equals("INSERT") && item.queryParam() != null && !item.queryParam().isEmpty()){
-                String dbKey = dbUtil.generateCacheKey(dbConfig);
+                String dbKey = dbUtil.generateCacheKey(dbConfig); // 存在当前内存缓存的key
                 String oldDataKey = dbKey.concat(tableName).concat((item.queryCriteria()).toString())
                                 .concat(item.queryParam().toString());
                 if(oldDataCache.asMap().putIfAbsent(oldDataKey, true) == null) {//添加成功
@@ -489,7 +493,7 @@ public class DbParamsService {
             this.event = event;
         }
         public String getEvent() {return event;}
-        private volatile String deleteColumns;
+        private volatile String deleteColumns; // String deleteColumns = "delete from tableName";
         public void setDeleteColumnss(String deleteColumns) {
             this.deleteColumns = deleteColumns;
         }
@@ -500,23 +504,23 @@ public class DbParamsService {
         public void setPage(Integer page) { this.page.set(page); }
         public int getPageSize() { return pageSize.get(); }
         public void setPageSize(Integer pageSize) { this.pageSize.set(pageSize);}
-        private volatile String permission;//行权限
+        private volatile String permission;//行权限   + "and dept = 1"
         public String getPermission() { return permission; }
         public void setPermission(String permission) { this.permission = permission; }
-        private volatile String selectColumns;//查询字段表达式
+        private volatile String selectColumns;//查询字段表达式  "select col1,col2... from tableName" || "select  * from tableName" 没有权限就是*
         public String getSelectColumns() { return selectColumns; }
         public void setSelectColumns(String selectColumns) { this.selectColumns = selectColumns; }
-        private volatile String insertColumns;//新增字段表达式
-        private final List<List<Object>> insertValues = new CopyOnWriteArrayList<>();//新增执行参数
-        private final Map<String, Object> addColumns = new HashMap<>();
+        private volatile String insertColumns;//新增字段表达式 "insert tableName(col1,col2,...) values (?,?,...) "
+        private final List<List<Object>> insertValues = new CopyOnWriteArrayList<>(); //新增执行参数 ?的值,通过顺序对应?
+        private final Map<String, Object> addColumns = new HashMap<>(); //  字段名 = 字段值  col1 = 1,col2 = "1"
         public String getInsertColumns() { return insertColumns; }
         public void setInsertColumns(String insertColumns) { this.insertColumns = insertColumns; }
         public List<List<Object>> getInsertValues() { return insertValues; }
         public void setInsertValues(List<List<Object>> insertValues) { this.insertValues.addAll(insertValues); }
         public void setAddColumns(Map<String, Object> addColumns) { this.addColumns.putAll(addColumns); }
         public Map<String, Object> getAddColumns() { return addColumns; }
-        private volatile String updateColumns;//更新字段表达式
-        private final Map<String, List<List<Object>>> queryConditions = new ConcurrentHashMap<>();//查询参数集合
+        private volatile String updateColumns;//更新字段表达式 // update tableName set col1 = ? , col2=? ,...
+        private final Map<String, List<List<Object>>> queryConditions = new ConcurrentHashMap<>();//查询参数集合 "where  col1 = ? and col2 = ?...":[value1,value2...]
         public String getUpdateColumns() { return updateColumns; }
         public void setUpdateColumns(String updateColumns) { this.updateColumns = updateColumns; }
         public Map<String, List<List<Object>>> getQueryConditions() { return queryConditions; }

src/main/java/com/bfkj/unidia/DataUtils/ScriptExecutor.java

@@ -87,6 +87,10 @@ public class ScriptExecutor {
             injectParams(context, params);// 将参数注入JS上下文（支持嵌套Map结构）
             Value resultValue = executeScript(context, script, scriptKey);// 执行JavaScript脚本并获取原始结果值
             Object resultData = convertResult(resultValue);// 将JS结果值转换为Java对象
+            if (resultData instanceof String){
+                logger.info("js方法执行:");
+                logger.info(String.valueOf(resultData));
+            }
             return Result.success(resultData, buildMetrics(startTime, script,params != null ? params.size() : 0));
         } catch (Exception e) {
             logger.error("JS执行异常: {}, 脚本: {}", e.getMessage(), script, e);
@@ -609,6 +613,12 @@ public class ScriptExecutor {
             try {
                 return Integer.parseInt(strValue);
             }catch (Exception e){
+                logger.error("value");
+                logger.error(String.valueOf(value));
+                logger.error("strValue");
+                logger.error(strValue);
+                logger.error("targetType",targetType);
+                logger.error(String.valueOf(targetType));
                 logger.error("转换Integer失败",e);
                 return null;
             }

src/main/java/com/bfkj/unidia/IOUtils/IBMMQHelper.java

@@ -5,6 +5,7 @@ import com.bfkj.unidia.Result;
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.ibm.mq.*;
+import com.ibm.mq.constants.CMQC;
 import com.ibm.mq.constants.MQConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -12,9 +13,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PreDestroy;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 @Component
 public class IBMMQHelper {
@@ -182,6 +181,8 @@ public class IBMMQHelper {
     public static Result<List<String>> receiveMessage(Map<String, String> mqConfig,
                                                       String queueName,
                                                       Integer maxMessages) {
+        logger.info("IBM MQ配置-------------------------------------------");
+        logger.info(DataFormatConverter.mapToJson(mqConfig));
         MQQueueManager queueManager = null;
         String cacheKey = null;
         try {
@@ -203,8 +204,95 @@ public class IBMMQHelper {
                 if(queue == null){
                     return Result.fail("绑定队列失败");
                 }
+
+
+                List<String> messages = new ArrayList<>();
+                //创建消息获取选项并设置等待时间
+                MQGetMessageOptions gmo = new MQGetMessageOptions();
+                gmo.options = MQConstants.MQGMO_WAIT | MQConstants.MQGMO_FAIL_IF_QUIESCING;
+                gmo.waitInterval = 5000;
+                //gmo.options = MQConstants.MQGMO_WAIT | MQConstants.MQGMO_CONVERT | MQConstants.MQGMO_FAIL_IF_QUIESCING;
+                //gmo.waitInterval = 0; // 0表示无限等
+//                //不能获取当前队列深度
+//                int currentDepth = queue.getCurrentDepth();
+//                if (currentDepth == 0) {//队列深度为0，返回空结果
+//                    return Result.success(new ArrayList<>());
+//                }
+                //for (int i = 0; i < Math.min(maxMessages, currentDepth); i++) {//循环获取指定数量的消息
+                for (int i = 0; i < maxMessages; i++) {//循环获取指定数量的消息
+                    try {
+                        MQMessage message = new MQMessage();
+                        try {
+                            queue.get(message, gmo);
+                            byte[] buffer = new byte[message.getMessageLength()];
+                            message.readFully(buffer);
+                            String read =  new String(buffer);
+                            logger.info("接收 IBM MQ 队列消息：{}", read);
+                            messages.add(read);
+                        } catch (MQException e) {
+                            if (e.getReason() == MQConstants.MQRC_NO_MSG_AVAILABLE) {
+                                break; // 没有消息是正常情况，跳出循环
+                            }
+                            throw e; // 其他MQ异常需要上层处理
+                        }
+                    } catch (MQException e) {
+                        if (e.reasonCode == MQConstants.MQRC_NO_MSG_AVAILABLE) {
+                            logger.info("IBMMQ没有更多消息");
+                            break; // 没有更多消息，提前结束
+                        } else {
+                            logger.error("接收 IBM MQ 消息失败", e);
+                            return Result.fail("接收消息失败: " + e.getMessage());
+                        }
+                    }
+                }
+                Result<List<String>> listResult = Result.success(messages);
+                logger.info("最大接收数量:"+maxMessages);
+                logger.info("队列名称:"+queueName);
+                if (Objects.nonNull(listResult.getData())&&listResult.getData().size()>0){
+                    logger.info("发送数量",listResult.getData().size());
+                }
+                return listResult;
+            } catch (Exception e) {
+                closeQueue(cacheKey, queue, consumerQueueCache);
+                logger.error("IBM MQ 接收异常", e);
+                return Result.fail("IBM MQ 收异常: " + e.getMessage());
+            }
+        } catch (Exception e) {
+            logger.error("IBM MQ 接收异常", e);
+            closeQueueManager(cacheKey,queueManager);
+            return Result.fail("IBM MQ 接收异常: " + e.getMessage());
+        }
+    }
+/*    public static Result<List<String>> receiveMessage(Map<String, String> mqConfig,
+                                                      String queueName,
+                                                      Integer maxMessages) {
+        MQQueueManager queueManager = null;
+        String cacheKey = null;
+        try {
+            //配置验证阶段
+            Result<String> validateResult = validateConfig(mqConfig, queueName, "data");
+            if (!validateResult.isSuccess()) {
+                return Result.fail(validateResult.getError());
+            }
+            cacheKey = validateResult.getData();
+            //获取队列管理器
+            queueManager = getQueueManager(mqConfig, cacheKey);
+            if (queueManager == null) {
+                return Result.fail("获取队列管理器失败");
+            }
+            MQQueue queue = null;
+            try {
+                //绑定队列
+                queue = getconsumerMQQueue(queueManager, cacheKey, queueName);
+                logger.info("*************************绑定队列深度");
+                logger.info("队列深度:"+queue);
+                if(queue == null){
+                    return Result.fail("绑定队列失败");
+                }
                 //获取当前队列深度
                 int currentDepth = queue.getCurrentDepth();
+                logger.info("获取队列深度");
+                logger.info("队列深度:"+currentDepth);
                 if (currentDepth == 0) {//队列深度为0，返回空结果
                     return Result.success(new ArrayList<>());
                 }
@@ -251,7 +339,7 @@ public class IBMMQHelper {
             closeQueueManager(cacheKey,queueManager);
             return Result.fail("IBM MQ 接收异常: " + e.getMessage());
         }
-    }
+    }*/
 
     /**
      * 获取或创建 IBM MQ 队列管理器实例
@@ -270,12 +358,18 @@ public class IBMMQHelper {
      * @param cacheKey  缓存键值，用于唯一标识该队列管理器实例
      * @return          成功时返回队列管理器实例，失败返回 null
      */
-    private static MQQueueManager getQueueManager(Map<String, String> mqConfig, String cacheKey) {
+/*    private static MQQueueManager getQueueManager(Map<String, String> mqConfig, String cacheKey) {
         // 一级缓存检查：无需同步的快速失败路径
         MQQueueManager queueManager = queueManagerCache.getIfPresent(cacheKey);
         if (queueManager != null && queueManager.isConnected()) {
             return queueManager;
         }
+        logger.info("********************获取或创建 IBM MQ 队列管理器实例");
+        if (!Objects.nonNull(mqConfig)){
+            if (Objects.nonNull(mqConfig.get("host"))) logger.info("host:"+mqConfig.get("host"));
+            if (Objects.nonNull(mqConfig.get("port"))) logger.info("端口:"+mqConfig.get("port"));
+        }
+
 
         // 同步块确保线程安全，防止重复创建连接
         synchronized (connectionLock) {
@@ -290,16 +384,20 @@ public class IBMMQHelper {
                 MQEnvironment.hostname = mqConfig.get("host");
                 MQEnvironment.port = Integer.parseInt(mqConfig.get("port"));
                 MQEnvironment.channel = mqConfig.get("channel");
-                MQEnvironment.userID = mqConfig.get("username");
-                MQEnvironment.password = mqConfig.get("password");
+                if (StringUtils.isNullOrEmpty(mqConfig.get("username"))){
+                    MQEnvironment.userID = mqConfig.get("username");
+                }
+                if (StringUtils.isNullOrEmpty(mqConfig.get("password"))){
+                    MQEnvironment.password = mqConfig.get("password");
+                }
                 MQEnvironment.CCSID = Integer.parseInt(mqConfig.getOrDefault("ccsid", "1208"));
                 System.setProperty("com.ibm.mq.transport.type", mqConfig.getOrDefault("transportType", "1"));
 
                 // 创建新的队列管理器实例
-                queueManager = new MQQueueManager(mqConfig.get("queueManager"));
-
+ps -ef
                 // 将新创建的实例存入缓存
                 queueManagerCache.put(cacheKey, queueManager);
+                logger.info("********************获取或创建 IBM MQ 队列管理器实例 end");
                 return queueManager;
             } catch (Exception e) {
                 // 记录连接失败的异常信息
@@ -307,8 +405,55 @@ public class IBMMQHelper {
                 return null;
             }
         }
-    }
+    }*/
+    private static MQQueueManager getQueueManager(Map<String, String> mqConfig, String cacheKey) {
+        // 一级缓存检查：无需同步的快速失败路径
+        MQQueueManager queueManager = queueManagerCache.getIfPresent(cacheKey);
+        if (queueManager != null && queueManager.isConnected()) {
+            return queueManager;
+        }
 
+        // 同步块确保线程安全，防止重复创建连接
+        synchronized (connectionLock) {
+            // 二级缓存检查：防止重复初始化
+            queueManager = queueManagerCache.getIfPresent(cacheKey);
+            if (queueManager != null && queueManager.isConnected()) {
+                return queueManager;
+            }
+
+            try {
+                //
+                Properties props = new Properties();
+                props.put(CMQC.HOST_NAME_PROPERTY, mqConfig.get("host"));
+                props.put(CMQC.PORT_PROPERTY, Integer.parseInt(mqConfig.get("port")));
+                props.put(CMQC.CHANNEL_PROPERTY, mqConfig.get("channel"));
+                props.put(CMQC.CCSID_PROPERTY,mqConfig.getOrDefault("ccsid", "1208"));
+
+                String userId = mqConfig.get("username");
+                if(userId != null && !userId.isEmpty()){
+                    props.put(CMQC.USER_ID_PROPERTY, userId);
+                }
+                String  password = mqConfig.get("password");
+                if(password != null && !password.isEmpty()) {
+                    props.put(CMQC.PASSWORD_PROPERTY, password);
+                }
+                String queueManagerName = mqConfig.get("queueManager");
+                System.setProperty("com.ibm.mq.transport.type", mqConfig.getOrDefault("transportType", "1"));
+                if(queueManagerName != null && !queueManagerName.isEmpty()) {
+                    queueManager = new MQQueueManager(queueManagerName, props);
+                    queueManagerCache.put(cacheKey, queueManager);
+                    return queueManager;
+                }else {
+                    logger.error("创建MQ队列管理器失败: 队列名为空{}", mqConfig);
+                    return null;
+                }
+            } catch (Exception e) {
+                // 记录连接失败的异常信息
+                logger.error("创建 IBM MQ 连接失败", e);
+                return null;
+            }
+        }
+    }
     /**
      * 获取或创建一个MQ队列实例用于生产者操作。
      * 该方法首先尝试从缓存中获取已存在的队列实例，若队列不存在或未打开，

src/main/java/com/bfkj/unidia/IOUtils/KafkaHelper.java

@@ -70,13 +70,16 @@ public class KafkaHelper {
                 return Result.fail(producerResult.getError());
             }
             Producer<String, String> producer = producerResult.getData();
-
+            Result<Integer> integerResult = null;
             // 处理不同数据类型的消息发送
             if (data instanceof List<?> dataList) {
-                return batchSend(producer, topic, dataList,  cacheKey);
+                integerResult = batchSend(producer, topic, dataList, cacheKey);
             } else {
-                return sendSingleMessage(producer, topic, data,  cacheKey);
+                integerResult = sendSingleMessage(producer, topic, data, cacheKey);
             }
+            logger.info("Kafka发送消息返回结果:");
+            logger.info(integerResult.toString());
+            return integerResult;
         } catch (Exception e) {
             // 记录发送异常日志并返回错误结果
             logger.error("Kafka发送异常", e);
@@ -324,6 +327,9 @@ public class KafkaHelper {
                     return Result.fail("Kafka消费者{".concat(cacheKey).concat("}异常"));
                 }
             }
+            logger.info("消费数量:");
+            logger.info(String.valueOf(messages.size()));
+            if (messages.size()>0) logger.info(messages.get(0));
             return Result.success(messages);
         }catch (Exception e) {
             closeConsumer(cacheKey,consumer);

src/main/java/com/bfkj/unidia/OpenApi/DataProcessApi.java

@@ -82,5 +82,6 @@ public class DataProcessApi {
             return ResponseEntity.status(500).body(Result.fail("服务失败: " + e.getMessage()));
         }
     }
+
 }
 

src/main/java/com/bfkj/unidia/SystemEnvInitializer.java

@@ -237,6 +237,7 @@ public class SystemEnvInitializer implements ApplicationListener<ApplicationRead
         // 创建容器信息结构，包含数据体和唯一查询条件
         Map<String, Object> containerInfo = new HashMap<>() ;
         containerInfo.put("data",systemInfo);
+        // 条件: 用于判断是否存在, 存在修改,不存在新增
         containerInfo.put("conditions", Map.of("container_code", containerCode));
 
         // 组装最终请求参数

src/main/resources/application.properties

@@ -13,20 +13,32 @@ app.directories.plugins=plugins
 app.directories.database=db
 #????????????
 app.directories.temp=tmp
+# 人大金仓
+#spring.datasource.driverClassName=com.kingbase8.Driver
+#spring.datasource.url=jdbc:kingbase8://10.36.7.81:54321/service
+#spring.datasource.username=system
+#spring.datasource.password=12345678ab
+
+#spring.datasource.driverClassName=com.kingbase8.Driver
+#spring.datasource.url=jdbc:kingbase8://192.168.52.130:54321/base
+#spring.datasource.username=system
+#spring.datasource.password=system
 
 # SQLite ?????
-spring.datasource.url=jdbc:sqlite:file:D:\\UNIDIA206\\UNIDIA\\UNIDIA\\db\\unidia206.db   # ????????????????????
-spring.datasource.username=  # SQLite ?????
-spring.datasource.password=  # SQLite ????
+#spring.datasource.url=jdbc:sqlite:file:D:\\UNIDIA206\\UNIDIA\\UNIDIA\\db\\unidia206.db   # ????????????????????
+spring.datasource.url=jdbc:sqlite:file:G:\\WorkSpace\\UNIDIA\\db\\unidia206.db   # ????????????????????
+#spring.datasource.url=jdbc:sqlite:/app/service/db/unidia206.db
+spring.datasource.username=
+spring.datasource.password=
 spring.datasource.driver-class-name=org.sqlite.JDBC
-
-#spring.datasource.url=jdbc:mysql://192.168.243.11:3306/UNIDIA?useSSL=false&serverTimezone=UTC&characterEncoding=utf8
+# 国航
+#spring.datasource.url=jdbc:mysql://10.211.67.155:3306/UNIDIA?useSSL=false&serverTimezone=UTC&characterEncoding=utf8
 #spring.datasource.username=bigdata
-#spring.datasource.password=ENC(AES:IV:mgYR4Iixb2P8t1iBRf73sg==)
+#spring.datasource.password=QtrmuqDw^bJu$
 #spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
 # ????????????
 heartbeat.interval=2
 # ??????
 heartbeat.max-retries=3
 # ??????????
-heartbeat.retry-interval=1000
+heartbeat.retry-interval=1000