1、优化性能可读性

fs-qw-api/src/main/java/com/fs/app/service/impl/OpenQwApiServiceImpl.java

@@ -87,234 +87,220 @@ public class OpenQwApiServiceImpl implements OpenQwApiService {
     @Autowired
     private TenantDataSourceUtil tenantDataSourceUtil;
 
+    /** 同步用户线程池 */
+    private static final ExecutorService SYNC_USER_EXECUTOR = new ThreadPoolExecutor(
+            8, 16, 60L, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(500),
+            new ThreadPoolExecutor.CallerRunsPolicy()
+    );
+
     @Override
     public R getSyncQwUser(Long tenantId, String corpId) {
         String key = "qw:sync:" + corpId;
 
-        // 检查是否正在同步
         if (redisCache.hasKey(key)) {
             return R.error("同步任务正在执行中，请稍后再试");
         }
 
-        // 设置锁，防止重复（使用你的方法签名）
         boolean locked = redisCache.setIfAbsent(key, String.valueOf(System.currentTimeMillis()), 300, TimeUnit.SECONDS);
         if (!locked) {
             return R.error("同步任务正在执行中，请稍后再试");
         }
 
-        // 异步执行同步任务
-        asyncExecuteSync(tenantId, corpId, key);
+        // 异步执行同步任务（使用线程池代替@Async自调用失效问题）
+        SYNC_USER_EXECUTOR.submit(() -> {
+            try {
+                executeSync(tenantId, corpId);
+            } finally {
+                redisCache.deleteObject(key);
+            }
+        });
 
-        // 立即返回
         return R.ok("同步任务已启动，请稍后查看结果");
     }
 
-    /**
-     * 异步执行同步逻辑
-     */
-    @Async
-    public void asyncExecuteSync(Long tenantId, String corpId, String lockKey) {
-        try {
-            executeSync(tenantId, corpId);
-        } finally {
-            // 同步完成后删除锁
-            redisCache.deleteObject(lockKey);
-        }
-    }
-
     /**
      * 实际的同步逻辑
      */
     private void executeSync(Long tenantId, String corpId) {
         tenantDataSourceUtil.executeWithResult(tenantId, () -> {
             long startTime = System.currentTimeMillis();
-            log.info("========== 开始同步用户数据 ==========");
-            log.info("租户ID: {}, 企业ID: {}", tenantId, corpId);
+            log.info("========== 开始同步用户数据 ========== 租户ID: {}, 企业ID: {}", tenantId, corpId);
 
             try {
-                // ========== 1. 获取部门列表 ==========
-                log.info("步骤1: 获取部门列表");
-                QwDeptResult deptResult = qwApiService.getDepartmentList(corpId);
-                List<Department> departmentList = deptResult.getDepartment();
-
-                if (departmentList == null || departmentList.isEmpty()) {
-                    log.warn("未获取到任何部门，同步结束");
+                // 1. 获取部门列表
+                List<Department> departmentList = fetchDepartments(corpId);
+                if (departmentList == null) {
                     return R.ok("未获取到部门列表");
                 }
-                log.info("获取到部门数量: {}", departmentList.size());
 
-                // ========== 2. 获取企业信息和access_token ==========
-                log.info("步骤2: 获取企业信息和access_token");
+                // 2. 获取access_token
                 QwCompany qwCompany = iQwCompanyService.selectQwCompanyByCorpId(corpId);
                 if (qwCompany == null) {
                     log.error("未找到企业信息, corpId: {}", corpId);
                     return R.error("未找到企业信息");
                 }
                 String accessToken = qwApiService.getToken(corpId, qwCompany.getPermanentCode());
-                log.info("获取access_token成功");
-
-                // ========== 3. 批量获取所有部门的用户（去重） ==========
-                log.info("步骤3: 遍历所有部门获取用户列表");
-                Map<String, DeptUserResult> userMap = new ConcurrentHashMap<>();
-                int totalDeptUsers = 0;
-                int emptyDeptCount = 0;
-
-                for (Department department : departmentList) {
-                    try {
-                        log.debug("正在获取部门 [{}] {} 的用户列表", department.getId(), department.getName());
-                        UserResult userResult = qwApiService.getUserSimpleList(corpId, accessToken, department.getId());
-                        List<DeptUserResult> deptUsers = userResult.getUserlist();
-
-                        if (deptUsers == null || deptUsers.isEmpty()) {
-                            log.debug("部门 [{}] {} 没有用户", department.getId(), department.getName());
-                            emptyDeptCount++;
-                            continue;
-                        }
-
-                        // 使用putIfAbsent实现去重（保留第一次出现的用户信息）
-                        for (DeptUserResult user : deptUsers) {
-                            userMap.putIfAbsent(user.getUserid(), user);
-                        }
-                        totalDeptUsers += deptUsers.size();
-
-                    } catch (Exception e) {
-                        log.error("获取部门 [{}] {} 的用户列表失败", department.getId(), department.getName(), e);
-                    }
-                }
-
-                log.info("部门统计: 总部门数={}, 空部门数={}, 原始用户总数={}, 去重后用户数={}",
-                        departmentList.size(), emptyDeptCount, totalDeptUsers, userMap.size());
 
+                // 3. 并发获取所有部门用户（去重）
+                Map<String, DeptUserResult> userMap = fetchDeptUsersConcurrently(corpId, accessToken, departmentList);
                 if (userMap.isEmpty()) {
-                    log.info("无用户需要同步");
                     return R.ok("无用户需要同步");
                 }
 
-                // ========== 4. 查询数据库中已存在的用户 ==========
-                log.info("步骤4: 查询数据库中已存在的用户");
-                List<String> userIds = new ArrayList<>(userMap.keySet());
-                List<QwUser> existingUsers = qwUserMapper.selectQwUsersByCorpIdAndUserIds(corpId, userIds);
-                Map<String, QwUser> existingUserMap = existingUsers.stream()
-                        .collect(Collectors.toMap(QwUser::getQwOpenUserId, Function.identity()));
-
-                log.info("数据库已存在用户数: {}, 新增用户数: {}",
-                        existingUserMap.size(), userIds.size() - existingUserMap.size());
-
-                // ========== 5. 批量获取用户详细信息并构建处理对象 ==========
-                log.info("步骤5: 批量处理用户详情（每批100个）");
-                List<QwUser> usersToProcess = new ArrayList<>();
-                List<List<String>> batches = Lists.partition(userIds, 100);
-
-                int successCount = 0;
-                int errorCount = 0;
-
-                for (int batchIndex = 0; batchIndex < batches.size(); batchIndex++) {
-                    List<String> batch = batches.get(batchIndex);
-                    log.info("处理第 {}/{} 批，本批用户数: {}", batchIndex + 1, batches.size(), batch.size());
-
-                    List<QwUser> batchUsers = new ArrayList<>();
-                    for (String userId : batch) {
-                        try {
-                            DeptUserResult apiUser = userMap.get(userId);
-                            if (apiUser == null) {
-                                log.warn("用户 {} 在API结果中不存在，跳过", userId);
-                                errorCount++;
-                                continue;
-                            }
-
-                            QwUser existingQwUser = existingUserMap.get(userId);
-                            boolean isNewUser = (existingQwUser == null);
+                // 4. 查询数据库中已存在的用户
+                Map<String, QwUser> existingUserMap = queryExistingUsers(corpId, userMap.keySet());
 
-                            // 调用API转换openid
-                            QwOpenidByUserParams params = new QwOpenidByUserParams();
-                            params.setUserid(userId);
-                            QwOpenidResult openidResult = qwApiService.useridToOpenid(params, corpId);
+                // 5. 并发转换openid并构建用户对象
+                List<QwUser> usersToProcess = convertUsersConcurrently(corpId, userMap, existingUserMap);
 
-                            // 构建QwUser对象
-                            QwUser qwUser = new QwUser();
+                // 6. 批量数据库操作
+                int errorCount = (int) (userMap.size() - usersToProcess.size());
+                saveUsersToDatabase(usersToProcess, errorCount, startTime);
 
-                            // 设置部门（取第一个部门）
-                            List<Integer> departmentList_ = apiUser.getDepartment();
-                            if (departmentList_ != null && !departmentList_.isEmpty()) {
-                                qwUser.setDepartment(String.valueOf(departmentList_.get(0)));
-                            } else {
-                                qwUser.setDepartment("");
-                            }
+                return R.ok("同步完成");
 
-                            qwUser.setQwUserName(apiUser.getName());
-                            qwUser.setCorpId(corpId);
-                            qwUser.setOpenid(openidResult.getOpenid());
-                            qwUser.setQwOpenUserId(userId);
+            } catch (Exception e) {
+                log.error("同步用户过程发生异常", e);
+                return R.error("同步失败：" + e.getMessage());
+            }
+        });
+    }
 
-                            // 设置id（存在则设置，不存在则为null）
-                            if (existingQwUser != null) {
-                                qwUser.setId(existingQwUser.getId());
-                            } else {
-                                qwUser.setId(null);
-                            }
+    /**
+     * 获取部门列表
+     */
+    private List<Department> fetchDepartments(String corpId) {
+        QwDeptResult deptResult = qwApiService.getDepartmentList(corpId);
+        List<Department> departmentList = deptResult.getDepartment();
+        if (departmentList == null || departmentList.isEmpty()) {
+            log.warn("未获取到任何部门，同步结束");
+            return null;
+        }
+        log.info("获取到部门数量: {}", departmentList.size());
+        return departmentList;
+    }
 
-                            batchUsers.add(qwUser);
-                            successCount++;
+    /**
+     * 并发获取所有部门用户并去重
+     */
+    private Map<String, DeptUserResult> fetchDeptUsersConcurrently(String corpId, String accessToken, List<Department> departmentList) {
+        Map<String, DeptUserResult> userMap = new ConcurrentHashMap<>();
 
-                        } catch (Exception e) {
-                            log.error("处理用户失败，userId: {}", userId, e);
-                            errorCount++;
+        List<CompletableFuture<Void>> futures = departmentList.stream()
+                .map(dept -> CompletableFuture.runAsync(() -> {
+                    try {
+                        UserResult userResult = qwApiService.getUserSimpleList(corpId, accessToken, dept.getId());
+                        List<DeptUserResult> deptUsers = userResult.getUserlist();
+                        if (deptUsers != null && !deptUsers.isEmpty()) {
+                            deptUsers.forEach(user -> userMap.putIfAbsent(user.getUserid(), user));
                         }
+                    } catch (Exception e) {
+                        log.error("获取部门 [{}] {} 的用户列表失败", dept.getId(), dept.getName(), e);
                     }
+                }, SYNC_USER_EXECUTOR))
+                .collect(Collectors.toList());
 
-                    usersToProcess.addAll(batchUsers);
-                    log.info("第 {} 批处理完成: 成功={}, 失败={}", batchIndex + 1, batchUsers.size(), batch.size() - batchUsers.size());
-                }
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
 
-                log.info("用户处理统计: 成功={}, 失败={}", successCount, errorCount);
-                log.info("最终待处理用户数: {}", usersToProcess.size());
+        log.info("部门用户去重后数量: {}", userMap.size());
+        return userMap;
+    }
 
-                if (usersToProcess.isEmpty()) {
-                    log.info("没有需要新增或更新的用户");
-                    return R.ok(String.format("同步完成: 失败%d个", errorCount));
-                }
+    /**
+     * 查询数据库中已存在的用户
+     */
+    private Map<String, QwUser> queryExistingUsers(String corpId, Set<String> userIds) {
+        List<QwUser> existingUsers = qwUserMapper.selectQwUsersByCorpIdAndUserIds(corpId, new ArrayList<>(userIds));
+        Map<String, QwUser> existingUserMap = existingUsers.stream()
+                .collect(Collectors.toMap(QwUser::getQwOpenUserId, Function.identity()));
+        log.info("数据库已存在用户数: {}, 新增用户数: {}", existingUserMap.size(), userIds.size() - existingUserMap.size());
+        return existingUserMap;
+    }
 
-                // ========== 6. 批量数据库操作 ==========
-                log.info("步骤6: 执行数据库批量操作");
-                List<QwUser> toInsert = usersToProcess.stream()
-                        .filter(u -> u.getId() == null)
-                        .collect(Collectors.toList());
-                List<QwUser> toUpdate = usersToProcess.stream()
-                        .filter(u -> u.getId() != null)
-                        .collect(Collectors.toList());
+    /**
+     * 并发转换openid并构建QwUser对象
+     */
+    private List<QwUser> convertUsersConcurrently(String corpId, Map<String, DeptUserResult> userMap, Map<String, QwUser> existingUserMap) {
+        AtomicInteger errorCount = new AtomicInteger(0);
 
-                log.info("数据库操作: 待新增{}条, 待更新{}条", toInsert.size(), toUpdate.size());
+        List<CompletableFuture<QwUser>> futures = userMap.entrySet().stream()
+                .map(entry -> CompletableFuture.supplyAsync(() -> {
+                    String userId = entry.getKey();
+                    DeptUserResult apiUser = entry.getValue();
+                    try {
+                        // 调用API转换openid
+                        QwOpenidByUserParams params = new QwOpenidByUserParams();
+                        params.setUserid(userId);
+                        QwOpenidResult openidResult = qwApiService.useridToOpenid(params, corpId);
 
-                int insertSuccess = 0;
-                int updateSuccess = 0;
+                        return buildQwUser(apiUser, existingUserMap.get(userId), corpId, openidResult.getOpenid());
+                    } catch (Exception e) {
+                        log.error("处理用户失败，userId: {}", userId, e);
+                        errorCount.incrementAndGet();
+                        return null;
+                    }
+                }, SYNC_USER_EXECUTOR))
+                .collect(Collectors.toList());
 
-                if (!toInsert.isEmpty()) {
-                    insertSuccess = qwUserMapper.batchUpdateQwUser(toInsert);
-                    log.info("批量新增用户成功: {}条", insertSuccess);
-                }
+        List<QwUser> result = futures.stream()
+                .map(CompletableFuture::join)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toList());
 
-                if (!toUpdate.isEmpty()) {
-                    updateSuccess = qwUserMapper.batchUpdateQwUser(toUpdate);
-                    log.info("批量更新用户成功: {}条", updateSuccess);
-                }
+        log.info("用户处理统计: 成功={}, 失败={}", result.size(), errorCount.get());
+        return result;
+    }
+
+    /**
+     * 构建QwUser对象
+     */
+    private QwUser buildQwUser(DeptUserResult apiUser, QwUser existingUser, String corpId, String openid) {
+        QwUser qwUser = new QwUser();
+        qwUser.setQwUserName(apiUser.getName());
+        qwUser.setCorpId(corpId);
+        qwUser.setIsDel(0);
+        qwUser.setOpenid(openid);
+        qwUser.setQwOpenUserId(apiUser.getUserid());
+
+        // 设置部门（取第一个部门）
+        List<Integer> depts = apiUser.getDepartment();
+        qwUser.setDepartment(depts != null && !depts.isEmpty() ? String.valueOf(depts.get(0)) : "");
+
+        // 存在则设置id（更新），不存在则为null（新增）
+        if (existingUser != null) {
+            qwUser.setId(existingUser.getId());
+        }
+        return qwUser;
+    }
+
+    /**
+     * 批量保存用户到数据库
+     */
+    private void saveUsersToDatabase(List<QwUser> users, int errorCount, long startTime) {
+        if (users.isEmpty()) {
+            log.info("没有需要新增或更新的用户");
+            return;
+        }
 
-                long endTime = System.currentTimeMillis();
-                String resultMsg = String.format(
-                        "同步完成！总耗时: %d ms | 新增: %d/%d | 更新: %d/%d | 失败: %d",
-                        (endTime - startTime), insertSuccess, toInsert.size(),
-                        updateSuccess, toUpdate.size(), errorCount
-                );
+        List<QwUser> toInsert = users.stream().filter(u -> u.getId() == null).collect(Collectors.toList());
+        List<QwUser> toUpdate = users.stream().filter(u -> u.getId() != null).collect(Collectors.toList());
 
-                log.info(resultMsg);
-                log.info("========== 用户同步结束 ==========");
+        log.info("数据库操作: 待新增{}条, 待更新{}条", toInsert.size(), toUpdate.size());
 
-                return R.ok(resultMsg);
+        int insertSuccess = 0, updateSuccess = 0;
+        if (!toInsert.isEmpty()) {
+            insertSuccess = qwUserMapper.batchUpdateQwUser(toInsert);
+        }
+        if (!toUpdate.isEmpty()) {
+            updateSuccess = qwUserMapper.batchUpdateQwUser(toUpdate);
+        }
 
-            } catch (Exception e) {
-                log.error("同步用户过程发生异常", e);
-                return R.error("同步失败：" + e.getMessage());
-            }
-        });
+        long elapsed = System.currentTimeMillis() - startTime;
+        String resultMsg = String.format("同步完成！总耗时: %d ms | 新增: %d/%d | 更新: %d/%d | 失败: %d",
+                elapsed, insertSuccess, toInsert.size(), updateSuccess, toUpdate.size(), errorCount);
+        log.info(resultMsg);
+        log.info("========== 用户同步结束 ==========");
     }
 
     /**

fs-service/src/main/resources/mapper/qw/QwUserMapper.xml

@@ -384,7 +384,6 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         qw_user_name = VALUES(qw_user_name),
         department = VALUES(department),
         openid = VALUES(openid),
-        is_del = VALUES(is_del),
         update_time = NOW()
     </insert>
 </mapper>