|
|
@@ -54,6 +54,7 @@ import java.time.format.DateTimeFormatter;
|
|
|
import java.util.*;
|
|
|
import java.util.concurrent.*;
|
|
|
import java.util.function.Function;
|
|
|
+import java.util.stream.Collector;
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
/**
|
|
|
@@ -744,12 +745,44 @@ public class QwSopLogsServiceImpl implements IQwSopLogsService
|
|
|
List<QwSopLogsDoSendListTVO> logsByJsApi = qwSopLogsMapper.getQwSopLogsByJsApiAll(param);
|
|
|
|
|
|
|
|
|
+// // 优先返回 sendType == 8 的第一条记录
|
|
|
+// List<QwSopLogsDoSendListTVO> result = logsByJsApi.stream()
|
|
|
+// .filter(log -> log.getSendType() == 8)
|
|
|
+// .findFirst()
|
|
|
+// .map(Collections::singletonList) // 单元素不可变 List
|
|
|
+// .orElse(logsByJsApi); // 如果没有匹配项，返回原列表
|
|
|
// 优先返回 sendType == 8 的第一条记录
|
|
|
List<QwSopLogsDoSendListTVO> result = logsByJsApi.stream()
|
|
|
- .filter(log -> log.getSendType() == 8)
|
|
|
- .findFirst()
|
|
|
- .map(Collections::singletonList) // 单元素不可变 List
|
|
|
- .orElse(logsByJsApi); // 如果没有匹配项，返回原列表
|
|
|
+ .collect(Collector.of(
|
|
|
+ () -> new Object() {
|
|
|
+ QwSopLogsDoSendListTVO type8 = null;
|
|
|
+ List<QwSopLogsDoSendListTVO> type9List = new ArrayList<>();
|
|
|
+ List<QwSopLogsDoSendListTVO> otherList = new ArrayList<>();
|
|
|
+ },
|
|
|
+ (acc, log) -> {
|
|
|
+ if (acc.type8 == null && log.getSendType() == 8) {
|
|
|
+ acc.type8 = log;
|
|
|
+ } else if (acc.type8 == null) {
|
|
|
+ if (log.getSendType() == 9) {
|
|
|
+ acc.type9List.add(log);
|
|
|
+ } else {
|
|
|
+ acc.otherList.add(log);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ (acc1, acc2) -> { throw new UnsupportedOperationException(); },
|
|
|
+ acc -> {
|
|
|
+ if (acc.type8 != null) {
|
|
|
+ return Collections.singletonList(acc.type8);
|
|
|
+ } else if (!acc.type9List.isEmpty()) {
|
|
|
+ List<QwSopLogsDoSendListTVO> combined = new ArrayList<>(acc.type9List.size() + acc.otherList.size());
|
|
|
+ combined.addAll(acc.type9List);
|
|
|
+ combined.addAll(acc.otherList);
|
|
|
+ return combined;
|
|
|
+ }
|
|
|
+ return logsByJsApi;
|
|
|
+ }
|
|
|
+ ));
|
|
|
|
|
|
// 查询员工信息的id
|
|
|
QwUser qwUser = qwExternalContactService.getQwUserByRedis(param.getCorpId().trim(),param.getQwUserId().trim());
|
|
|
@@ -1326,173 +1359,178 @@ public class QwSopLogsServiceImpl implements IQwSopLogsService
|
|
|
* 该方法负责处理企业微信的群发消息创建和发送
|
|
|
*/
|
|
|
@Override
|
|
|
- public void createCorpMassSending(String date) {
|
|
|
+ public void createCorpMassSending(String date) {
|
|
|
+
|
|
|
long startTime = System.currentTimeMillis();
|
|
|
logger.info("开始执行企业微信群发消息创建任务");
|
|
|
|
|
|
- String json = configService.selectConfigByKey("course.config");
|
|
|
- CourseConfig config = JSON.parseObject(json, CourseConfig.class);
|
|
|
-
|
|
|
- if (config == null) {
|
|
|
- logger.error("课程默认配置为空，不执行");
|
|
|
- return;
|
|
|
- }
|
|
|
- // 获取需要发送的SOP日志记录
|
|
|
- List<QwSopLogs> qwSopLogs = qwSopLogsMapper.selectSopLogsByCreateCorpMassSending(date);
|
|
|
-// List<QwSopLogs> qwSopLogs = qwSopLogsMapper.checkQwSopLogs();
|
|
|
+ List<QwSopLogs> qwSopLogs = qwSopLogsMapper.createCorpMassSending(date);
|
|
|
if (qwSopLogs.isEmpty()) {
|
|
|
logger.error("zyp \n【企微官方群发记录为空】");
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
- // 按照企业员工ID、发送时间、SOP ID和企业ID进行分组
|
|
|
Map<String, List<QwSopLogs>> groupedLogs = new HashMap<>();
|
|
|
for (QwSopLogs log : qwSopLogs) {
|
|
|
String key = log.getQwUserid() + "|" + log.getSendTime() + "|" + log.getSopId() + "|" + log.getCorpId();
|
|
|
groupedLogs.computeIfAbsent(key, k -> new ArrayList<>()).add(log);
|
|
|
}
|
|
|
|
|
|
- // 创建线程池，使用固定大小的线程池以避免过多线程导致的资源竞争
|
|
|
int threadCount = Math.min(10, Runtime.getRuntime().availableProcessors() + 1);
|
|
|
ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
|
|
|
-
|
|
|
- // 创建用于发送消息的嵌套线程池
|
|
|
ExecutorService messageExecutorService = Executors.newFixedThreadPool(20);
|
|
|
-
|
|
|
- // 用于存储需要批量更新的日志记录，使用线程安全的集合
|
|
|
List<QwSopLogs> updateList = Collections.synchronizedList(new ArrayList<>());
|
|
|
|
|
|
- // 使用CountDownLatch等待所有任务完成
|
|
|
CountDownLatch latch = new CountDownLatch(groupedLogs.size());
|
|
|
|
|
|
- // 处理每个分组
|
|
|
- for (Map.Entry<String, List<QwSopLogs>> entry : groupedLogs.entrySet()) {
|
|
|
- String key = entry.getKey();
|
|
|
- List<QwSopLogs> logs = entry.getValue();
|
|
|
- String[] keys = key.split("\\|");
|
|
|
- String qwUserid = keys[0];
|
|
|
- String corpId = keys[3];
|
|
|
-
|
|
|
-// QwUser qwUser = qwUserMapper.selectQwUserByCorpIdAndUserId(corpId, qwUserid);
|
|
|
- // 查询员工信息的id
|
|
|
- QwUser qwUser = qwExternalContactService.getQwUserByRedis(corpId.trim(),qwUserid.trim());
|
|
|
- if (qwUser != null && qwUser.getIsDel() == 0) {
|
|
|
- // 提交到线程池处理每个分组
|
|
|
- executorService.submit(() -> {
|
|
|
- try {
|
|
|
- // 按外部用户ID分组
|
|
|
- Map<String, List<QwSopLogs>> userLogsMap = new HashMap<>();
|
|
|
- for (QwSopLogs log : logs) {
|
|
|
- String externalUserId = log.getExternalUserId();
|
|
|
- userLogsMap.computeIfAbsent(externalUserId, k -> new ArrayList<>()).add(log);
|
|
|
- }
|
|
|
-
|
|
|
- // 使用嵌套的CountDownLatch等待所有消息发送完成
|
|
|
- CountDownLatch messageLatch = new CountDownLatch(userLogsMap.size());
|
|
|
-
|
|
|
- // 处理每个外部用户
|
|
|
- for (Map.Entry<String, List<QwSopLogs>> userEntry : userLogsMap.entrySet()) {
|
|
|
- String externalUserId = userEntry.getKey();
|
|
|
- List<QwSopLogs> userLogs = userEntry.getValue();
|
|
|
+ try {
|
|
|
+ for (Map.Entry<String, List<QwSopLogs>> entry : groupedLogs.entrySet()) {
|
|
|
+ String key = entry.getKey();
|
|
|
+ List<QwSopLogs> logs = entry.getValue();
|
|
|
+ String[] keys = key.split("\\|");
|
|
|
+ String qwUserid = keys[0];
|
|
|
+ String corpId = keys[3];
|
|
|
+
|
|
|
+ QwUser qwUser = qwExternalContactService.getQwUserByRedis(corpId.trim(), qwUserid.trim());
|
|
|
+ if (qwUser != null && qwUser.getIsDel() == 0) {
|
|
|
+ executorService.submit(() -> {
|
|
|
+ try {
|
|
|
+ Map<String, List<QwSopLogs>> userLogsMap = new HashMap<>();
|
|
|
+ for (QwSopLogs log : logs) {
|
|
|
+ userLogsMap.computeIfAbsent(log.getExternalUserId(), k -> new ArrayList<>()).add(log);
|
|
|
+ }
|
|
|
|
|
|
- // 提交到消息发送线程池
|
|
|
- messageExecutorService.submit(() -> {
|
|
|
- try {
|
|
|
- QwMsgTemplateSop templateSop = new QwMsgTemplateSop();
|
|
|
- templateSop.setChatType("single");
|
|
|
- templateSop.setAllowSelect(false);
|
|
|
- templateSop.setSender(qwUserid);
|
|
|
- templateSop.setExternalUseridList(Collections.singletonList(externalUserId));
|
|
|
-
|
|
|
- List<QwMsgTemplateSop.Attachment> attachments = new ArrayList<>();
|
|
|
- boolean hasError = false;
|
|
|
-
|
|
|
- for (QwSopLogs log : userLogs) {
|
|
|
- try {
|
|
|
- QwSopTempSetting.Content content = JSON.parseObject(log.getContentJson(), QwSopTempSetting.Content.class);
|
|
|
- if (content == null || content.getSetting() == null) continue;
|
|
|
- Long courseId = content.getCourseId();
|
|
|
- for (QwSopTempSetting.Content.Setting set : content.getSetting()) {
|
|
|
- processContent(set, corpId, templateSop, attachments, courseId);
|
|
|
+ CountDownLatch messageLatch = new CountDownLatch(userLogsMap.size());
|
|
|
+ for (Map.Entry<String, List<QwSopLogs>> userEntry : userLogsMap.entrySet()) {
|
|
|
+ String externalUserId = userEntry.getKey();
|
|
|
+ List<QwSopLogs> userLogs = userEntry.getValue();
|
|
|
+
|
|
|
+ messageExecutorService.submit(() -> {
|
|
|
+ try {
|
|
|
+ QwMsgTemplateSop templateSop = new QwMsgTemplateSop();
|
|
|
+ templateSop.setChatType("single");
|
|
|
+ templateSop.setAllowSelect(false);
|
|
|
+ templateSop.setSender(qwUserid);
|
|
|
+ templateSop.setExternalUseridList(Collections.singletonList(externalUserId));
|
|
|
+
|
|
|
+ List<QwMsgTemplateSop.Attachment> attachments = new ArrayList<>();
|
|
|
+ boolean hasError = false;
|
|
|
+ for (QwSopLogs log : userLogs) {
|
|
|
+ try {
|
|
|
+ QwSopTempSetting.Content content = JSON.parseObject(log.getContentJson(), QwSopTempSetting.Content.class);
|
|
|
+ if (content == null || content.getSetting() == null) continue;
|
|
|
+ Long courseId = content.getCourseId();
|
|
|
+ for (QwSopTempSetting.Content.Setting set : content.getSetting()) {
|
|
|
+ processContent(set, corpId, templateSop, attachments, courseId);
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ logger.error("消息内容解析失败，logId：{},{},{}", log.getId(), e,key);
|
|
|
+ hasError = true;
|
|
|
}
|
|
|
- } catch (Exception e) {
|
|
|
- logger.error("消息内容解析失败，logId：{}", log.getId(), e);
|
|
|
- hasError = true;
|
|
|
}
|
|
|
- }
|
|
|
+ if (!hasError && (!attachments.isEmpty() || templateSop.getTextContent() != null)) {
|
|
|
+ templateSop.setAttachments(attachments);
|
|
|
+ try {
|
|
|
+ QwAddMsgTemplateResult result = qwApiService.addMsgTemplateBySop(templateSop, corpId);
|
|
|
+ if (result.getErrCode() == 0 || result.getErrCode() == 41063){
|
|
|
+ for (QwSopLogs log : userLogs) {
|
|
|
+ log.setSendStatus(1L);
|
|
|
+ log.setMsgId(result.getMsgId());
|
|
|
+ updateList.add(log);
|
|
|
+ }
|
|
|
+ }else {
|
|
|
+
|
|
|
+ for (QwSopLogs log : userLogs) {
|
|
|
+ log.setSendType(2);
|
|
|
+ log.setSendStatus(3L);
|
|
|
+ log.setRemark("官方有误,sop补发");
|
|
|
+ log.setReceivingStatus(0L);
|
|
|
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+ LocalDateTime currentTime = LocalDateTime.now();
|
|
|
+ String newTimeString = currentTime.format(formatter);
|
|
|
+ log.setSendTime(newTimeString);
|
|
|
+ log.setSort(30000001);
|
|
|
+ updateList.add(log);
|
|
|
+ }
|
|
|
|
|
|
- if (!hasError && (!attachments.isEmpty() || templateSop.getTextContent() != null)) {
|
|
|
- templateSop.setAttachments(attachments);
|
|
|
- try {
|
|
|
- QwAddMsgTemplateResult result = qwApiService.addMsgTemplateBySop(templateSop, corpId);
|
|
|
- for (QwSopLogs log : userLogs) {
|
|
|
- log.setSendStatus(result.getErrCode() == 0 || result.getErrCode() == 41063 ? 1L : 0L);
|
|
|
- log.setMsgId(result.getMsgId());
|
|
|
- updateList.add(log);
|
|
|
- }
|
|
|
- if (result.getErrCode() != 0 && result.getErrCode() != 41063){
|
|
|
- logger.error("企业微信接口-消息发送失败，corpId：{}，errCode：{}，errMsg：{}", corpId, result.getErrCode(), result.getErrMsg());
|
|
|
- }
|
|
|
- } catch (Exception e) {
|
|
|
- logger.error("消息发送失败，user：{}", externalUserId, e);
|
|
|
- for (QwSopLogs log : userLogs) {
|
|
|
- log.setSendStatus(0L);
|
|
|
- updateList.add(log);
|
|
|
+ logger.error("企业微信接口-消息发送失败-进入sop补偿，corpId：{}，errCode：{}，errMsg：{},key:{}", corpId, result.getErrCode(), result.getErrMsg(),key);
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ logger.error("消息发送失败，user：{},{},{}", externalUserId, e,key);
|
|
|
+ for (QwSopLogs log : userLogs) {
|
|
|
+ log.setSendStatus(0L);
|
|
|
+ log.setRemark("信息异常");
|
|
|
+ updateList.add(log);
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+ } finally {
|
|
|
+ logger.info("执行结束-messageLatch-countDown："+updateList.size());
|
|
|
+ messageLatch.countDown();
|
|
|
}
|
|
|
- } finally {
|
|
|
- messageLatch.countDown();
|
|
|
- }
|
|
|
- });
|
|
|
- }
|
|
|
+ });
|
|
|
|
|
|
- // 等待所有消息发送完成
|
|
|
- try {
|
|
|
+ }
|
|
|
+
|
|
|
+ logger.info("messageExecutorService-updateList总量："+updateList.size());
|
|
|
+
|
|
|
+ // 等待所有消息发送完成
|
|
|
messageLatch.await();
|
|
|
+
|
|
|
} catch (InterruptedException e) {
|
|
|
- logger.error("等待消息发送完成时被中断", e);
|
|
|
+ logger.info("messageExecutorService-Thread.currentThread().interrupt()："+updateList.size());
|
|
|
Thread.currentThread().interrupt();
|
|
|
+ } finally {
|
|
|
+ logger.info("finally-latch.countDown:"+updateList.size());
|
|
|
+ latch.countDown();
|
|
|
}
|
|
|
- } finally {
|
|
|
- latch.countDown();
|
|
|
- }
|
|
|
- });
|
|
|
- }else {
|
|
|
- logger.error("官方群发 员工信息有误："+corpId+":"+qwUserid);
|
|
|
+ });
|
|
|
+ } else {
|
|
|
+ logger.error("员工信息无效-不存在或被删除，corpId:{}, userId:{}", corpId, qwUserid);
|
|
|
+
|
|
|
+ latch.countDown(); // 确保每个分组都减少计数
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- }
|
|
|
+ latch.await(); // 等待所有分组提交的任务完成
|
|
|
+ logger.info("关闭线程池并等待任务完成："+updateList.size());
|
|
|
+ // 关闭线程池并等待任务完成
|
|
|
+ executorService.shutdown();
|
|
|
+ messageExecutorService.shutdown();
|
|
|
+ if (!executorService.awaitTermination(300, TimeUnit.SECONDS)) {
|
|
|
+ logger.error("ExecutorService未完全关闭");
|
|
|
+ }
|
|
|
+ if (!messageExecutorService.awaitTermination(300, TimeUnit.SECONDS)) {
|
|
|
+ logger.error("MessageExecutorService未完全关闭");
|
|
|
+ }
|
|
|
|
|
|
- // 等待所有分组处理完成
|
|
|
- try {
|
|
|
- latch.await();
|
|
|
- } catch (InterruptedException e) {
|
|
|
- logger.error("等待分组处理完成时被中断", e);
|
|
|
- Thread.currentThread().interrupt();
|
|
|
- }
|
|
|
+ // 5. 同步块生成快照（终极防护），创建快照避免并发修改
|
|
|
+ List<QwSopLogs> batchList;
|
|
|
+ synchronized (updateList) { // 加锁确保无并发修改
|
|
|
+ batchList = new ArrayList<>(updateList);
|
|
|
+ }
|
|
|
|
|
|
- // 批量更新发送状态，每500条一批
|
|
|
- if (!updateList.isEmpty()) {
|
|
|
- int batchSize = 500;
|
|
|
- for (int i = 0; i < updateList.size(); i += batchSize) {
|
|
|
- int endIndex = Math.min(i + batchSize, updateList.size());
|
|
|
- List<QwSopLogs> batch = updateList.subList(i, endIndex);
|
|
|
- try {
|
|
|
- qwSopLogsMapper.batchUpdateStatus(batch);
|
|
|
- logger.info("批量修改 sopLogs 成功，修改数量: " + batch.size());
|
|
|
- } catch (Exception e) {
|
|
|
- logger.error("批量修改 sopLogs 失败", e);
|
|
|
+
|
|
|
+ logger.info("批量修改总数: {}", batchList.size());
|
|
|
+
|
|
|
+ if (!batchList.isEmpty()){
|
|
|
+ int batchSize = 1000;
|
|
|
+ for (int i = 0; i < batchList.size(); i += batchSize) {
|
|
|
+ int end = Math.min(i + batchSize, batchList.size());
|
|
|
+ List<QwSopLogs> subList = batchList.subList(i, end);
|
|
|
+ qwSopLogsMapper.batchUpdateStatus(subList);
|
|
|
}
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- // 关闭线程池
|
|
|
- executorService.shutdown();
|
|
|
- messageExecutorService.shutdown();
|
|
|
|
|
|
- long endTime = System.currentTimeMillis();
|
|
|
- logger.info("企业微信群发消息创建任务执行完成，总耗时: {} 毫秒", (endTime - startTime));
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ logger.error("线程中断异常", e);
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ } finally {
|
|
|
+ long endTime = System.currentTimeMillis();
|
|
|
+ logger.info("任务完成，耗时: {} 毫秒", endTime - startTime);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
// 处理不同类型的内容
|
|
|
@@ -1593,12 +1631,12 @@ public class QwSopLogsServiceImpl implements IQwSopLogsService
|
|
|
// }
|
|
|
|
|
|
@Override
|
|
|
- public void createCorpMassSendingByUserLogs(String date) {
|
|
|
+ public void createCorpMassSendingByUserLogs(String taskStartTime,String taskEndTime) {
|
|
|
|
|
|
long startTime = System.currentTimeMillis();
|
|
|
logger.info("开始执行企业微信群发消息创建任务");
|
|
|
|
|
|
- List<QwSopLogs> qwSopLogsList = qwSopLogsMapper.selectSopLogsByCreateCorpMassSending(date);
|
|
|
+ List<QwSopLogs> qwSopLogsList = qwSopLogsMapper.selectSopLogsByCreateCorpMassSending(taskStartTime,taskEndTime);
|
|
|
if (qwSopLogsList.isEmpty()) {
|
|
|
logger.error("zyp \n【企微官方群发记录为空】");
|
|
|
return;
|
