canalclient优化

canal-console/pom.xml

@@ -37,18 +37,6 @@
             <artifactId>ruoyi-framework</artifactId>
         </dependency>
 
-        <!-- 定时任务-->
-        <dependency>
-            <groupId>com.ruoyi</groupId>
-            <artifactId>ruoyi-quartz</artifactId>
-        </dependency>
-
-        <!-- 代码生成-->
-        <dependency>
-            <groupId>com.ruoyi</groupId>
-            <artifactId>ruoyi-generator</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
@@ -58,6 +46,12 @@
         <dependency>
             <groupId>com.retdata</groupId>
             <artifactId>yaoyi-bi</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.ruoyi</groupId>
+                    <artifactId>ruoyi-quartz</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
     </dependencies>

canal-console/src/main/java/com/retdata/canal/CannalClient.java

@@ -25,10 +25,15 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 @Component
-public class CannalClient implements InitializingBean {
+public class CannalClient {
+
+    static {
+        log.info("CanalClient类已加载");
+    }
 
     @Autowired
     private CanalConfig canalConfig;
@@ -41,39 +46,86 @@ public class CannalClient implements InitializingBean {
 
     private Map<String, AbstractCanalDbHandler> tableNameHandlerMap = new HashMap<>();
 
-    private final static int BATCH_SIZE = 1000;
+    private final static int BATCH_SIZE = 500;
 
     private final static long LOCAL_THREAD_SLEEP = 2000L;
 
-    private Map<DataSourceEnum, CanalConnector> canalConnectors= new HashMap<>(2);
+    private Map<DataSourceEnum, CanalConnector> canalConnectors= new HashMap<>(DataSourceEnum.values().length);
+    private final Map<DataSourceEnum, Thread> canalThreads = new HashMap<>(DataSourceEnum.values().length);
 
     @PostConstruct
     public void initialize() {
+        log.info("CanalClient开始初始化，注册的处理器数量: {}", canalDbHandlerMap.size());
         for (Map.Entry<String, AbstractCanalDbHandler> entry : canalDbHandlerMap.entrySet()) {
             tableNameHandlerMap.put(entry.getValue().tableName(), entry.getValue());
         }
+
+        try {
+            start();
+        }catch (Exception e){
+            log.error("CanalClient初始化异常", e);
+        }
+
+        log.info("CanalClient初始化完成");
     }
 
     @PreDestroy
     public void stop(){
-        if(CollUtil.isNotEmpty(canalConnectors)){
-            for(Map.Entry<DataSourceEnum, CanalConnector> entry: canalConnectors.entrySet()){
-                CanalConnector canalConnector = entry.getValue();
-                if (canalConnector instanceof ClusterCanalConnector) {
-                    ((ClusterCanalConnector) canalConnector).stopRunning();
-                } else if (canalConnector instanceof SimpleCanalConnector) {
-                    ((SimpleCanalConnector) canalConnector).stopRunning();
+        log.info("开始停止CanalClient...");
+
+        // 1. 设置停止标志
+        running = false;
+
+        // 2. 等待当前正在处理的数据完成
+        log.info("等待当前数据处理完成...");
+        try {
+            // 给更多时间让当前数据处理完成
+            Thread.sleep(10000); // 10秒
+        } catch (InterruptedException e) {
+            log.warn("等待数据处理完成时被中断", e);
+        }
+
+        // 3. 关闭Canal连接
+        log.info("开始关闭Canal连接...");
+        if (CollUtil.isNotEmpty(canalConnectors)) {
+            for (Map.Entry<DataSourceEnum, CanalConnector> entry : canalConnectors.entrySet()) {
+                try {
+                    CanalConnector canalConnector = entry.getValue();
+                    if (canalConnector instanceof ClusterCanalConnector) {
+                        ((ClusterCanalConnector) canalConnector).stopRunning();
+                    } else if (canalConnector instanceof SimpleCanalConnector) {
+                        ((SimpleCanalConnector) canalConnector).stopRunning();
+                    }
+                    log.info("Canal连接已关闭: {}", entry.getKey().name());
+                } catch (Exception e) {
+                    log.error("关闭Canal连接时发生异常: {}", entry.getKey().name(), e);
                 }
             }
         }
 
-        running = false;
-        executor.shutdownNow();
-        log.warn("CannalClient 已销毁");
+        // 4. 优雅关闭线程池
+        log.info("开始关闭线程池...");
+        try {
+            executor.shutdown();
+            // 等待线程池中的任务完成，最多等待30秒
+            if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
+                log.warn("线程池未能在30秒内完成，强制关闭");
+                executor.shutdownNow();
+                // 再等待5秒
+                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+                    log.error("线程池无法关闭");
+                }
+            }
+        } catch (InterruptedException e) {
+            log.error("关闭线程池时发生异常", e);
+            executor.shutdownNow();
+        }
+
+        log.warn("CanalClient 已销毁");
     }
 
-    @Override
-    public void afterPropertiesSet() throws Exception {
+    public void start() throws Exception {
+        log.info("CanalClient开始启动，配置信息: {}", canalConfig);
         if (canalConfig != null && MapUtil.isNotEmpty(canalConfig.getSourceConfig())) {
             for (DataSourceEnum dataSourceEnum : canalConfig.getSourceConfig().keySet()) {
                 CanalConfig.SourceConfig sourceConfig = canalConfig.getSourceConfig().get(dataSourceEnum);
@@ -82,66 +134,108 @@ public class CannalClient implements InitializingBean {
                 }
 
                 executor.submit(()-> {
-                    while (running){
-                        try {
-                            while (true) {
-                                // 创建链接
-                                CanalConnector connector = null;
-                                if(CollUtil.isNotEmpty(canalConnectors)
-                                        && canalConnectors.containsKey(dataSourceEnum)
-                                ){
-                                    connector = canalConnectors.get(dataSourceEnum);
-                                }else {
-                                    connector = CanalConnectors.newSingleConnector(new InetSocketAddress(sourceConfig.getServerHost(), sourceConfig.getServerPort()), sourceConfig.getInstance(), sourceConfig.getUsername(), sourceConfig.getPassword());
-
-                                    canalConnectors.put(dataSourceEnum, connector);
-                                }
-                                try {
-                                    //打开连接
-                                    connector.connect();
-                                    //订阅数据库表,全部表
-                                    connector.subscribe(sourceConfig.getSubscribe());
-                                    //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
-                                    connector.rollback();
-                                    while (true) {
-                                        // 获取指定数量的数据
-                                        Message message = connector.getWithoutAck(BATCH_SIZE);
-                                        //获取批量ID
-                                        long batchId = message.getId();
-                                        //获取批量的数量
-                                        int size = message.getEntries().size();
-
-                                        //如果没有数据
-                                        if (batchId == -1 || size == 0) {
-                                            try {
-                                                //线程休眠2秒
-                                                Thread.sleep(LOCAL_THREAD_SLEEP);
-                                            } catch (InterruptedException e) {
-                                                e.printStackTrace();
+                    Thread currentThread = Thread.currentThread();
+                    canalThreads.put(dataSourceEnum, currentThread);
+
+                    try {
+                        log.info("Canal线程启动，数据源: {}, 线程ID: {}, 线程名: {}",
+                                dataSourceEnum.name(),
+                                currentThread.getId(),
+                                currentThread.getName());
+                        while (running) {
+                            try {
+                                while (running) {
+                                    // 创建链接
+                                    CanalConnector connector = null;
+                                    if (CollUtil.isNotEmpty(canalConnectors)
+                                            && canalConnectors.containsKey(dataSourceEnum)
+                                    ) {
+                                        connector = canalConnectors.get(dataSourceEnum);
+                                    } else {
+                                        connector = CanalConnectors.newSingleConnector(new InetSocketAddress(sourceConfig.getServerHost(), sourceConfig.getServerPort()), sourceConfig.getInstance(), sourceConfig.getUsername(), sourceConfig.getPassword());
+
+                                        canalConnectors.put(dataSourceEnum, connector);
+                                    }
+                                    try {
+                                        //打开连接
+                                        connector.connect();
+                                        //订阅数据库表,全部表
+                                        connector.subscribe(sourceConfig.getSubscribe());
+                                        //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
+                                        connector.rollback();
+
+                                        while (true) {
+                                            // 检查应用是否正在关闭
+                                            if (!running) {
+                                                log.info("应用正在关闭，停止获取数据，数据源: {}", dataSourceEnum.name());
+                                                break;
                                             }
-                                        } else {
-                                            //如果有数据,处理数据
-                                            writeOut(dataSourceEnum, message);
+
+                                            // 获取指定数量的数据
+                                            Message message = connector.getWithoutAck(BATCH_SIZE);
+                                            //获取批量ID
+                                            long batchId = message.getId();
+                                            //获取批量的数量
+                                            int size = message.getEntries().size();
+
+                                            //如果没有数据
+                                            if (batchId == -1 || size == 0) {
+                                                try {
+                                                    log.warn("没有数据");
+                                                    //线程休眠2秒
+                                                    Thread.sleep(LOCAL_THREAD_SLEEP);
+                                                } catch (InterruptedException e) {
+                                                    e.printStackTrace();
+                                                }
+                                            } else {
+                                                log.info("数据源:{},数据:{}", dataSourceEnum, size);
+                                                //如果有数据,处理数据
+                                                writeOut(dataSourceEnum, message);
+                                            }
+                                            log.info("connector.ack({});", batchId);
+                                            //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
+                                            connector.ack(batchId);
                                         }
-                                        //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
-                                        connector.ack(batchId);
+                                    } catch (OutOfMemoryError e) {
+                                        log.error("内存不足错误，数据源: {}", dataSourceEnum.name(), e);
+                                        // 内存不足，可能需要重启应用
+                                        break;
+                                    } catch (StackOverflowError e) {
+                                        log.error("栈溢出错误，数据源: {}", dataSourceEnum.name(), e);
+                                        break;
+                                    } catch (ThreadDeath e) {
+                                        log.error("线程被强制终止，数据源: {}", dataSourceEnum.name(), e);
+                                        break;
+                                    } catch (Exception e) {
+                                        log.error("CanalClient应用程序异常，数据源: {}", dataSourceEnum.name(), e);
+                                        // 应用程序异常，可以尝试恢复
+                                    } catch (Error e) {
+                                        log.error("CanalClient系统级错误，数据源: {}", dataSourceEnum.name(), e);
+                                        // 系统级错误，重新抛出
+                                        throw e;
+                                    } finally {
+                                        connector.disconnect();
+                                        log.info("{} 连接已断开", dataSourceEnum);
                                     }
-                                } catch (Exception e) {
-                                    e.printStackTrace();
-                                } finally {
-                                    connector.disconnect();
-                                    DynamicDataSourceContextHolder.clearDataSourceType();
-                                }
 
-                                try {
-                                    Thread.sleep(sourceConfig.getInterval());
-                                } catch (InterruptedException e) {
-                                    throw new RuntimeException(e);
+                                    try {
+                                        Thread.sleep(sourceConfig.getInterval());
+                                    } catch (InterruptedException e) {
+                                        throw new RuntimeException(e);
+                                    }
                                 }
+                            } catch (Exception e) {
+                                log.error("CanalClient error", e);
                             }
-                        } catch (Exception e) {
-                            log.error("CanalClient error", e);
                         }
+                    } catch (Exception e) {
+                        log.error("Canal线程异常退出，数据源: {}, 线程ID: {}, 异常: {}",
+                                dataSourceEnum.name(),
+                                currentThread.getId(),
+                                e.getMessage(),
+                                e);
+                    } finally {
+                        canalThreads.remove(dataSourceEnum);
                     }
                 });
             }
@@ -152,9 +246,18 @@ public class CannalClient implements InitializingBean {
      * 打印canal server解析binlog获得的实体类信息
      */
     private void writeOut(DataSourceEnum dataSourceEnum, Message message) {
+        // 检查应用是否正在关闭
+        if (!running) {
+            log.warn("应用正在关闭，跳过数据处理");
+            return;
+        }
         List<CanalEntry.Entry> entrys = message.getEntries();
-        for (CanalEntry.Entry entry : entrys) {
-            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
+        log.info("writeOut entrys size:{}", entrys.size());
+
+        for (int i=0;i< entrys.size();i++) {
+            CanalEntry.Entry entry = entrys.get(i);
+            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
+                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                 //开启/关闭事务的实体类型，跳过
                 continue;
             }
@@ -169,6 +272,12 @@ public class CannalClient implements InitializingBean {
             //获取操作类型：insert/update/delete类型
             CanalEntry.EventType eventType = rowChage.getEventType();
 
+            if (eventType != CanalEntry.EventType.INSERT
+                    && eventType != CanalEntry.EventType.UPDATE
+                    && eventType != CanalEntry.EventType.DELETE) {
+                continue;
+            }
+
             //判断是否是DDL语句
             if (rowChage.getIsDdl()) {
                 continue;
@@ -179,19 +288,33 @@ public class CannalClient implements InitializingBean {
             for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                 AbstractCanalDbHandler canalDbHandler = tableNameHandlerMap.get(tableName);
                 if (canalDbHandler == null) {
-                    log.warn("{}没有对应的处理器", tableName);
+                    log.warn("{} 没有对应的处理器", tableName);
                     continue;
                 }
 
+                log.info("{} 开始处理", tableName);
                 try {
+                    // 再次检查应用状态
+                    if (!running) {
+                        log.warn("应用正在关闭，跳过处理: {}", tableName);
+                        break;
+                    }
                     DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.CSO.name());
                     canalDbHandler.processRow(rowData, eventType, dataSourceEnum);
                 }catch (Exception e){
-                    log.error("处理数据异常", e);
+                    if (!running) {
+                        log.warn("应用正在关闭，忽略异常: {}", tableName);
+                        break;
+                    } else {
+                        log.error("处理数据异常", e);
+                    }
                 } finally {
+                    log.info("{} 结束处理", tableName);
+
                     DynamicDataSourceContextHolder.clearDataSourceType();
                 }
             }
         }
+        log.info("writeOut 处理完成");
     }
 }

canal-console/src/main/java/com/retdata/canal/handler/ContractedProductChangeHandler.java

@@ -36,7 +36,9 @@ public class ContractedProductChangeHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<ContractedProductChange> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(ContractedProductChange::getId, valId).eq(ContractedProductChange::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(ContractedProductChange::getId, valId)
+                    .eq(ContractedProductChange::getDbSourceId, valSource);
             contractedProductChangeService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysDeptHandler.java

@@ -46,7 +46,9 @@ public class SysDeptHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<CsoSysDept> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(CsoSysDept::getDeptId, valId).eq(CsoSysDept::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(CsoSysDept::getDeptId, valId)
+                    .eq(CsoSysDept::getDbSourceId, valSource);
             csoSysDeptService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysDeptSubHandler.java

@@ -47,7 +47,9 @@ public class SysDeptSubHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<SysDeptSub> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(SysDeptSub::getSubId, valId).eq(SysDeptSub::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(SysDeptSub::getSubId, valId)
+                    .eq(SysDeptSub::getDbSourceId, valSource);
             sysDeptSubService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysDictItemHandler.java

@@ -36,7 +36,9 @@ public class SysDictItemHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<CsoSysDictItem> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(CsoSysDictItem::getId, valId).eq(CsoSysDictItem::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(CsoSysDictItem::getId, valId)
+                    .eq(CsoSysDictItem::getDbSourceId, valSource);
             csoSysDictItemService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
         ||eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysFinancialGmvSettleDataHandler.java

@@ -36,7 +36,9 @@ public class SysFinancialGmvSettleDataHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<SysFinancialGmvSettleData> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(SysFinancialGmvSettleData::getId, valId).eq(SysFinancialGmvSettleData::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(SysFinancialGmvSettleData::getId, valId)
+                    .eq(SysFinancialGmvSettleData::getDbSourceId, valSource);
             sysFinancialGmvSettleDataService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysImplPlanHandler.java

@@ -48,7 +48,9 @@ public class SysImplPlanHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<SysImplPlan> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(SysImplPlan::getPlanId, valId).eq(SysImplPlan::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(SysImplPlan::getPlanId, valId)
+                    .eq(SysImplPlan::getDbSourceId, valSource);
             sysImplPlanService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysRoleHandler.java

@@ -36,7 +36,9 @@ public class SysRoleHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<CsoSysRole> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(CsoSysRole::getRoleId, valId).eq(CsoSysRole::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(CsoSysRole::getRoleId, valId)
+                    .eq(CsoSysRole::getDbSourceId, valSource);
             csoSysRoleService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysUserHandler.java

@@ -48,7 +48,9 @@ public class SysUserHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<CsoSysUser> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(CsoSysUser::getUserId, valId).eq(CsoSysUser::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(CsoSysUser::getUserId, valId)
+                    .eq(CsoSysUser::getDbSourceId, valSource);
             csoSysUserService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/SysUserSubHandler.java

@@ -48,7 +48,9 @@ public class SysUserSubHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<SysUserSub> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(SysUserSub::getId, valId).eq(SysUserSub::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(SysUserSub::getId, valId)
+                    .eq(SysUserSub::getDbSourceId, valSource);
             sysUserSubService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmDaAgentHandler.java

@@ -47,7 +47,9 @@ public class WmDaAgentHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmDaAgent> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmDaAgent::getId, valId).eq(WmDaAgent::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmDaAgent::getId, valId)
+                    .eq(WmDaAgent::getDbSourceId, valSource);
             wmDaAgentService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmDaDrugEntDrugtableHandler.java

@@ -46,7 +46,9 @@ public class WmDaDrugEntDrugtableHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmDaDrugEntDrugtable> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmDaDrugEntDrugtable::getId, valId).eq(WmDaDrugEntDrugtable::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmDaDrugEntDrugtable::getId, valId)
+                    .eq(WmDaDrugEntDrugtable::getDbSourceId, valSource);
             wmDaDrugEntDrugtableService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmDaDrugEntHandler.java

@@ -47,7 +47,9 @@ public class WmDaDrugEntHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmDaDrugEnt> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmDaDrugEnt::getId, valId).eq(WmDaDrugEnt::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmDaDrugEnt::getId, valId)
+                    .eq(WmDaDrugEnt::getDbSourceId, valSource);
             wmDaDrugEntService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmScorePackageHandler.java

@@ -48,7 +48,9 @@ public class WmScorePackageHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmScorePackage> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmScorePackage::getId, valId).eq(WmScorePackage::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmScorePackage::getId, valId)
+                    .eq(WmScorePackage::getDbSourceId, valSource);
             wmScorePackageService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmScorePackageSettleNoteHandler.java

@@ -36,7 +36,9 @@ public class WmScorePackageSettleNoteHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<WmScorePackageSettleNote> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmScorePackageSettleNote::getId, valId).eq(WmScorePackageSettleNote::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmScorePackageSettleNote::getId, valId)
+                    .eq(WmScorePackageSettleNote::getDbSourceId, valSource);
             wmScorePackageSettleNoteService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmScorePackageStatusHandler.java

@@ -47,7 +47,9 @@ public class WmScorePackageStatusHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmScorePackageStatus> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmScorePackageStatus::getId, valId).eq(WmScorePackageStatus::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmScorePackageStatus::getId, valId)
+                    .eq(WmScorePackageStatus::getDbSourceId, valSource);
             wmScorePackageStatusService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/java/com/retdata/canal/handler/WmTaskHandler.java

@@ -48,7 +48,9 @@ public class WmTaskHandler extends AbstractCanalDbHandler {
             }
 
             LambdaQueryWrapper<WmTask> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmTask::getId, valId).eq(WmTask::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmTask::getId, valId)
+                    .eq(WmTask::getDbSourceId, valSource);
             wmTaskService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT

canal-console/src/main/java/com/retdata/canal/handler/WmTaskTypeHandler.java

@@ -79,7 +79,7 @@ public class WmTaskTypeHandler extends AbstractCanalDbHandler {
 
             LambdaQueryWrapper<WmTaskType> queryWrapper = new LambdaQueryWrapper<>();
             queryWrapper.eq(WmTaskType::getId, valId)
-                    .eq(WmTaskType::getId, valSource);
+                    .eq(WmTaskType::getDbSourceId, valSource);
 
             WmTaskType wmTaskType = JSON.parseObject(jsonStr, WmTaskType.class);
             wmTaskType.setDbSourceId(valSource);

canal-console/src/main/java/com/retdata/canal/handler/WmUserPlatQuizResHandler.java

@@ -36,7 +36,9 @@ public class WmUserPlatQuizResHandler extends AbstractCanalDbHandler {
                 }
             }
             LambdaQueryWrapper<WmUserPlatQuizRes> queryWrapper = new LambdaQueryWrapper<>();
-            queryWrapper.eq(WmUserPlatQuizRes::getResultId, valId).eq(WmUserPlatQuizRes::getDbSourceId, valSource);
+            queryWrapper
+                    .eq(WmUserPlatQuizRes::getResultId, valId)
+                    .eq(WmUserPlatQuizRes::getDbSourceId, valSource);
             wmUserPlatQuizResService.delete(queryWrapper);
         } else if (eventType == CanalEntry.EventType.UPDATE
                 || eventType == CanalEntry.EventType.INSERT) {

canal-console/src/main/resources/application.yml

@@ -40,6 +40,7 @@ logging:
     com.ruoyi: debug
     org.springframework: warn
     com.retdata.yaoyibi: debug
+    com.retdata.canal: info
 
 # 用户配置
 user:

ruoyi-admin/src/main/resources/application.yml

@@ -120,3 +120,21 @@ xss:
   excludes: /system/notice
   # 匹配链接
   urlPatterns: /system/*,/monitor/*,/tool/*
+
+canal:
+  sourceConfig:
+    CSO:
+      enabled: false
+      serverHost: localhost
+      serverPort: 11111
+      instance: example
+      subscribe: "cso_prod\\..*"
+      interval: 60000
+    LUOXIN_CSO:
+      enabled: false
+      serverHost: localhost
+      serverPort: 11111
+      instance: luoxincso
+      subscribe: "cso_prod\\..*"
+      interval: 60000
+

ruoyi-framework/src/main/java/com/ruoyi/framework/web/service/SysLoginService.java

@@ -64,7 +64,7 @@ public class SysLoginService
     public String login(String username, String password, String code, String uuid)
     {
         // 验证码校验
-        validateCaptcha(username, code, uuid);
+//        validateCaptcha(username, code, uuid);
         // 登录前置校验
         loginPreCheck(username, password);
         // 用户验证

yaoyi-bi/src/main/java/com/retdata/yaoyibi/domain/CsoSysDeptSummary.java

@@ -23,4 +23,6 @@ public class CsoSysDeptSummary {
      * 代理商类型，团队-AT, 公司-AE
      */
     private String agentType;
+
+    private Integer dbSourceId;
 }

yaoyi-bi/src/main/java/com/retdata/yaoyibi/service/impl/WmScorePackageServiceImpl.java

@@ -307,12 +307,12 @@ public class WmScorePackageServiceImpl extends ServiceImpl<WmScorePackageMapper,
     private List<String> getIdPath(WmScorePackageSearchDto dto) {
         List<String> ids = new ArrayList<>(3);
         if (StringUtils.hasText(dto.getGrandParentId())) {
-            ids.add(dto.getGrandParentId());
+            ids.add(dto.getGrandParentId()+"_"+dto.getDbSourceId());
         }
         if (StringUtils.hasText(dto.getRelationScoreId())) {
-            ids.add(dto.getRelationScoreId());
+            ids.add(dto.getRelationScoreId()+"_"+dto.getDbSourceId());
         }
-        ids.add(dto.getId().toString());
+        ids.add(dto.getId().toString()+"_"+dto.getDbSourceId());
         return ids;
     }
 
@@ -325,6 +325,7 @@ public class WmScorePackageServiceImpl extends ServiceImpl<WmScorePackageMapper,
     public XSSFWorkbook export(WmScorePackageSearchParams searchParams) {
         Map<PackageIdDto, WmScorePackageExportDto> exportDtoCache = new HashMap<>();
         List<WmScorePackageExportDto> exportDtoList = getBaseMapper().selectExportDto(searchParams);
+
         MultiValueMap<PackageIdDto, WmScorePackageExportDto> parentToChildren = new LinkedMultiValueMap<>();
         Set<PackageIdDto> matchedPackageIds = getBaseMapper().selectAllMatchedPackageIds(searchParams);
         for (WmScorePackageExportDto dto : exportDtoList) {

yaoyi-bi/src/main/resources/mapper/CsoSysDeptMapper.xml

@@ -64,16 +64,16 @@
         ) t3 on t3.dept_id=d1.dept_id
         left join (
             select pd1.send_package_dept_id,count(distinct ps.user_id) as package_user_count from
-            (select sp1.id,sp1.send_package_dept_id from wm_score_package sp1
+            (select sp1.id,sp1.send_package_dept_id, sp1.db_source_id from wm_score_package sp1
             union
-            select csp.id,sp2.send_package_dept_id from wm_score_package sp2
-            join wm_score_package csp on csp.relation_score_id=cast(sp2.id as nchar)
+            select csp.id,sp2.send_package_dept_id, sp2.db_source_id from wm_score_package sp2
+            join wm_score_package csp on csp.relation_score_id=cast(sp2.id as nchar) and sp2.db_source_id=csp.db_source_id
             union
-            select ccsp.id,sp3.send_package_dept_id from wm_score_package sp3
-            join wm_score_package csp2 on csp2.relation_score_id=cast(sp3.id as nchar)
-            join wm_score_package ccsp on ccsp.relation_score_id=cast(csp2.id as nchar)
+            select ccsp.id,sp3.send_package_dept_id, sp3.db_source_id from wm_score_package sp3
+            join wm_score_package csp2 on csp2.relation_score_id=cast(sp3.id as nchar) and sp3.db_source_id=csp2.db_source_id
+            join wm_score_package ccsp on ccsp.relation_score_id=cast(csp2.id as nchar) and csp2.db_source_id=ccsp.db_source_id
             ) as pd1
-            join wm_score_package_status ps on ps.package_id=cast(pd1.id as nchar) and ps.status=2
+            join wm_score_package_status ps on ps.package_id=cast(pd1.id as nchar) and ps.status=2 and ps.db_source_id=pd1.db_source_id
             group by pd1.send_package_dept_id
         ) t4 on t4.send_package_dept_id=d1.dept_id
     </sql>

yaoyi-bi/src/main/resources/mapper/WmScorePackageMapper.xml

@@ -5,6 +5,7 @@
 <mapper namespace="com.retdata.yaoyibi.mapper.WmScorePackageMapper">
     <resultMap id="searchDtoMap" type="com.retdata.yaoyibi.domain.WmScorePackageSearchDto">
         <id column="id" property="id"/>
+        <id column="db_source_id" property="dbSourceId"/>
         <result column="score_package_name" property="scorePackageName"/>
         <result column="typeid" property="typeid"/>
         <result column="score" property="score"/>
@@ -22,18 +23,20 @@
         <result column="level" property="level"/>
         <result column="root_package_send_package_dept_id" property="rootPackageSendPackageDeptId"/>
         <result column="settle_status" property="settleStatus"/>
-        <result column="db_source_id" property="dbSourceId"/>
         <association property="sendPackageDept" javaType="com.retdata.yaoyibi.domain.CsoSysDeptSummary">
             <id column="send_package_dept_dept_id" property="deptId"/>
             <result column="send_package_dept_name" property="name"/>
             <result column="send_package_dept_level" property="level"/>
+            <result column="db_source_id" property="dbSourceId"/>
         </association>
         <association property="dept" javaType="com.retdata.yaoyibi.domain.CsoSysDeptSummary">
             <id column="dept_dept_id" property="deptId"/>
+            <result column="db_source_id" property="dbSourceId"/>
         </association>
     </resultMap>
     <resultMap id="exportDtoMap" type="com.retdata.yaoyibi.domain.WmScorePackageExportDto">
         <id column="id" property="id"/>
+        <id column="db_source_id" property="dbSourceId"/>
         <result column="score_package_name" property="scorePackageName"/>
         <result column="typeid" property="typeid"/>
         <result column="score" property="score"/>
@@ -50,20 +53,22 @@
         <result column="p2p_amount" property="p2pAmount"/>
         <result column="p2p_service_amount" property="p2pServiceAmount"/>
         <result column="settle_status" property="settleStatus"/>
-        <result column="db_source_id" property="dbSourceId"/>
         <association property="sendPackageDept" javaType="com.retdata.yaoyibi.domain.CsoSysDeptSummary">
             <id column="send_package_dept_dept_id" property="deptId"/>
             <result column="send_package_dept_name" property="name"/>
             <result column="send_package_dept_level" property="level"/>
+            <result column="db_source_id" property="dbSourceId"/>
         </association>
         <association property="dept" javaType="com.retdata.yaoyibi.domain.CsoSysDeptSummary">
             <id column="dept_dept_id" property="deptId"/>
             <result column="dept_name" property="name"/>
             <result column="dept_level" property="level"/>
+            <result column="db_source_id" property="dbSourceId"/>
         </association>
     </resultMap>
     <resultMap id="settleTreeNodeMap" type="com.retdata.yaoyibi.domain.WmScorePackageSettleTreeNode">
         <id column="id" property="id"/>
+        <id column="db_source_id" property="dbSourceId"/>
         <result column="typeid" property="typeid"/>
         <result column="settle_amount" property="settleAmount"/>
         <result column="settle_actual_amount" property="settleActualAmount"/>
@@ -287,7 +292,7 @@
     <select id="selectSubPackagesByPackageIds" resultMap="searchDtoMap">
         select
             t1.id,t1.score_package_name,t1.score,t1.score_package_status,t1.create_time,t1.package_finish_time,t1.typeid,t1.relation_score_id,t1.p2p_amount,t1.p2p_service_amount,
-            s1.settle_amount,s1.actual_amount as settle_actual_amount,s1.notify_time as settle_notify_time,s1.settle_count
+            s1.settle_amount,s1.actual_amount as settle_actual_amount,s1.notify_time as settle_notify_time,s1.settle_count,t1.db_source_id
         from wm_score_package t1
         left join (
             select package_id,sum(settle_amount) as settle_amount,sum(actual_amount) as actual_amount,max(notify_time) as notify_time,count(1) as settle_count
@@ -447,49 +452,50 @@
         sum(case when dsp.sign_product='CSO1' then 1 else 0 end) as cso1Count,
         sum(case when dsp.sign_product='CSO2' then 1 else 0 end) as cso2Count
         from wm_score_package rp
-        left join (
-            select t2.dept_id,t1.sign_product
-            from contracted_product_change t1
-                inner join sys_dept t2 on t1.object_name COLLATE utf8mb4_general_ci=t2.name
-            where t1.sign_product in ('CSO1','CSO2') and not exists(select 1 from contracted_product_change t3 where t3.object_name=t1.object_name and t3.change_time>t1.change_time)
-        ) dsp on dsp.dept_id=rp.dept_id
-        where id in
-        (
-            select distinct COALESCE(case when t2.relation_score_id='' then null else t2.relation_score_id end,t2.id,t1.id) as root_package_id
-            from wm_score_package t1
+            inner join (
+                select distinct COALESCE(case when t2.relation_score_id='' then null else t2.relation_score_id end,t2.id,t1.id) as
+        root_package_id,t1.db_source_id
+                from wm_score_package t1
                 left join wm_score_package t2 on t1.relation_score_id=t2.id
-                    and t1.db_source_id=t2.db_source_id
-        <if test="(searchParams.rootPackageSendPackageDeptId!=null and searchParams.rootPackageSendPackageDeptId.length>0) or searchParams.rootPackageCreateTimeGe!=null or searchParams.rootPackageCreateTimeLe!=null">
-                left join wm_score_package t3 on t2.relation_score_id=t3.id
+                and t1.db_source_id=t2.db_source_id
+                <if test="(searchParams.rootPackageSendPackageDeptId!=null and searchParams.rootPackageSendPackageDeptId.length>0) or searchParams.rootPackageCreateTimeGe!=null or searchParams.rootPackageCreateTimeLe!=null">
+                    left join wm_score_package t3 on t2.relation_score_id=t3.id
                     and t2.db_source_id=t3.db_source_id
-        </if>
-        <if test="searchParams.settleNotifyTimeGe!=null or searchParams.settleNotifyTimeLe!=null">
-                left join (
+                </if>
+                <if test="searchParams.settleNotifyTimeGe!=null or searchParams.settleNotifyTimeLe!=null">
+                    left join (
                     select package_id,max(notify_time) as notify_time , db_source_id
                     from wm_score_package_settle_note
                     group by package_id, db_source_id
-                ) s1 on s1.package_id=t1.id
+                    ) s1 on s1.package_id=t1.id
                     and s1.db_source_id=t1.db_source_id
-        </if>
-        <if test="searchParams.receiverName!=null and searchParams.receiverName.length>0">
-            inner join (
-                select distinct t1.package_id,t1.db_source_id
-                from wm_score_package_status t1
+                </if>
+                <if test="searchParams.receiverName!=null and searchParams.receiverName.length>0">
+                    inner join (
+                    select distinct t1.package_id,t1.db_source_id
+                    from wm_score_package_status t1
                     inner join sys_user t2 on t1.user_id=t2.user_id
-                        and t1.db_source_id=t2.db_source_id
-                where t1.status=2 and t2.realname in
-                <foreach collection="searchParams.receiverName" item="item" separator="," open="(" close=")">#{item}</foreach>
-                union
-                select distinct t4.id as package_id,t4.db_source_id
-                from sys_dept t3
+                    and t1.db_source_id=t2.db_source_id
+                    where t1.status=2 and t2.realname in
+                    <foreach collection="searchParams.receiverName" item="item" separator="," open="(" close=")">#{item}</foreach>
+                    union
+                    select distinct t4.id as package_id,t4.db_source_id
+                    from sys_dept t3
                     inner join wm_score_package t4 on t3.dept_id=t4.dept_id and t4.typeid in ('0','1','2')
-                        and t3.db_source_id=t4.db_source_id
-                where t3.name in
-                <foreach collection="searchParams.receiverName" item="item" separator="," open="(" close=")">#{item}</foreach>
-            ) rnt on rnt.package_id=t1.id
-                and rnt.db_source_id=t1.db_source_id
-        </if>
-        <where><include refid="searchParamsSql"></include></where>)
+                    and t3.db_source_id=t4.db_source_id
+                    where t3.name in
+                    <foreach collection="searchParams.receiverName" item="item" separator="," open="(" close=")">#{item}</foreach>
+                    ) rnt on rnt.package_id=t1.id
+                    and rnt.db_source_id=t1.db_source_id
+                </if>
+                <where><include refid="searchParamsSql"></include></where>
+            ) rp1 on rp.id=rp1.root_package_id and rp.db_source_id=rp1.db_source_id
+        left join (
+            select t2.dept_id,t1.sign_product
+            from contracted_product_change t1
+                inner join sys_dept t2 on t1.object_name COLLATE utf8mb4_general_ci=t2.name and t1.db_source_id=t2.db_source_id
+            where t1.sign_product in ('CSO1','CSO2') and not exists(select 1 from contracted_product_change t3 where t3.object_name=t1.object_name and t3.change_time>t1.change_time)
+        ) dsp on dsp.dept_id=rp.dept_id
         group by
             <choose>
                 <when test="statUnit=='day'">
@@ -609,7 +615,7 @@
         select t1.id,t1.user_id,t1.package_id,t1.db_source_id
         from wm_score_package_status t1
             inner join <foreach collection="packageIds" item="item" separator="union all" open="(" close=")">select #{item.id} id</foreach> t2
-                on t1.package_id=t2.id
+                on t1.package_id=t2.id and t1.db_source_id=t2.db_source_id
         where t1.status=2 and t1.db_source_id=#{dbSourceId}
     </select>
     <select id="selectReceiverNames" resultType="string">