fix

midjourney/src/main/java/com/yhlxj/service/midjourney/impl/MidjourneyServiceImpl.java

@@ -14,6 +14,7 @@ import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.cyksj.common.exception.BusinessRuntimeException;
 import com.cyksj.common.task.GlobalThreadPoolTaskExecutor;
 import com.cyksj.common.util.*;
@@ -49,10 +50,7 @@ import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -113,6 +111,9 @@ public class MidjourneyServiceImpl implements MidjourneyService {
 
     private final MidjourneySessionMapper midjourneySessionMapper;
 
+    private static final Map<Long, Object> lockMap = new ConcurrentHashMap<>();
+
+
 
 
     @Override
@@ -1226,7 +1227,7 @@ public class MidjourneyServiceImpl implements MidjourneyService {
     @Override
     public synchronized void syscConversation() throws Exception {
         List<MidjourneyUserConversation> fastMidjourneyUserConversations = conversationMapper.selectList(Wrappers.lambdaQuery(MidjourneyUserConversation.class)
-                .notIn(MidjourneyUserConversation::getStatus, List.of("SUCCESS","MODAL","FAILURE")).orderByAsc(MidjourneyUserConversation::getId).last(" limit 100"));
+                .notIn(MidjourneyUserConversation::getStatus, List.of("SUCCESS","MODAL","FAILURE")).orderByAsc(MidjourneyUserConversation::getId).last(" limit 10"));
 
         JSONArray jsonArray = listByIds(fastMidjourneyUserConversations.stream().map(MidjourneyUserConversation::getTaskId).collect(Collectors.toList()));
         for (Object json : jsonArray) {
@@ -1289,56 +1290,77 @@ public class MidjourneyServiceImpl implements MidjourneyService {
 
     @Override
     public synchronized void syncPicture() {
-        int batchSize = 10;
+        int pageNum = 1;
+        int batchSize = 15;
         int maxRetries = 3;
-        while (true) {
+        int iterationCount = 0;
+        int iterationRetries = 3;
+        // 获取当前时间
+        //long currentTimeMillis = System.currentTimeMillis();
+        // 获取当前分钟，并计算下次任务执行的时间（下一个 5 分钟的 30 秒）
+        //long nextTaskTime = (currentTimeMillis / (5 * 60 * 1000)) * (5 * 60 * 1000) + (5 * 60 * 1000) + 30 * 1000;
+        while (iterationCount < iterationRetries) {
+            iterationCount++;
             // 查询需要上传的图片对话
-            List<MidjourneyUserConversation> imgConversations = conversationMapper.selectList(Wrappers.lambdaQuery(MidjourneyUserConversation.class)
+            List<MidjourneyUserConversation> imgConversations = conversationMapper.selectPage(new Page<>(pageNum, batchSize),Wrappers.lambdaQuery(MidjourneyUserConversation.class)
                     .eq(MidjourneyUserConversation::getStatus, "SUCCESS")
                     .notLike(MidjourneyUserConversation::getImageUrl, ".mj.galaxydvd.com")
                     .ge(MidjourneyUserConversation::getStartTime, DateUtil.offsetMinute(new Date(), -60 * 24 * 3).getTime())
                     .eq(MidjourneyUserConversation::getImageInvalid, Boolean.FALSE)
-                    .orderByAsc(MidjourneyUserConversation::getId)
-                    .last(" limit " + batchSize));
+                    .orderByAsc(MidjourneyUserConversation::getId)).getRecords();
 
             // 如果没有数据，跳出循环
             if (imgConversations.isEmpty()) {
                 log.info("所有图片已处理完毕，结束同步任务");
                 break;
             }
-
-            for (MidjourneyUserConversation conversation : imgConversations) {
-                String imageUrl = conversation.getImageUrl();
-                if (StringUtil.isBlank(imageUrl)) {
-                    log.warn("对话ID {} 的图片URL为空，跳过处理", conversation.getId());
-                    continue;
-                }
-
-                boolean uploadSuccess = false;
-                for (int i = 0; i < maxRetries; i++) {
-                    try {
-                        // 上传图片并更新URL
-                        String newImageUrl = uploadPic(imageUrl, "conversation" + conversation.getId());
-                        conversation.setImageUrl(newImageUrl);
-                        uploadSuccess = true;
-                        log.info("对话ID {} 图片上传成功，新的图片URL：{}", conversation.getId(), newImageUrl);
-                        break;
-                    } catch (IOException e) {
-                        log.error("对话ID {} 第 {} 次上传图片失败，原因：{}", conversation.getId(), i + 1, e.getMessage(), e);
+            TASK_EXECUTOR.execute(() -> {
+                for (MidjourneyUserConversation conversation : imgConversations) {
+                    String imageUrl = conversation.getImageUrl();
+                    if (StringUtil.isBlank(imageUrl)) {
+                        log.warn("对话ID {} 的图片URL为空，跳过处理", conversation.getId());
+                        continue;
+                    }
+                    String newImageUrl = StringUtils.EMPTY;
+                    boolean uploadSuccess = false;
+                    for (int i = 0; i < maxRetries; i++) {
+                        try {
+                            // 上传图片并更新URL
+                            newImageUrl = uploadPic(imageUrl, "conversation" + conversation.getId());
+                            uploadSuccess = true;
+                            log.info("对话ID {} 图片上传成功，新的图片URL：{}", conversation.getId(), newImageUrl);
+                            break;
+                        } catch (IOException e) {
+                            log.error("对话ID {} 第 {} 次上传图片失败，原因：{}", conversation.getId(), i + 1, e.getMessage(), e);
+                        }
                     }
+                    conversationMapper.update(null, Wrappers.lambdaUpdate(MidjourneyUserConversation.class)
+                            .set(StringUtils.isNotBlank(newImageUrl), MidjourneyUserConversation::getImageUrl, newImageUrl)
+                            .set(MidjourneyUserConversation::getImageInvalid, !uploadSuccess)
+                            .eq(MidjourneyUserConversation::getId, conversation.getId()));
                 }
-
-                conversation.setImageInvalid(!uploadSuccess);
-                conversationMapper.updateById(conversation);
-            }
-
+            });
             // 防止高频次查询，增加短暂的休眠时间
             try {
-                Thread.sleep(1000);
+                int sleepTime = imgConversations.size() < batchSize ? 10000 : 1000;
+                Thread.sleep(sleepTime);
             } catch (InterruptedException e) {
                 log.error("线程休眠被中断", e);
                 Thread.currentThread().interrupt();
             }
+            // 判断当前时间是否接近下次任务执行时间
+            //if (System.currentTimeMillis() > nextTaskTime - 30000) {
+            //    log.info("接近下次任务执行时间，提前结束同步任务");
+            //    break;
+            //}
+            // 增加页码，查询下一页数据
+            pageNum++;
+        }
+
+        if (iterationCount >= iterationRetries) {
+            log.warn("达到最大循环次数 {}，强制结束任务", iterationRetries);
+        } else {
+            log.info("图片同步任务提前结束，所有图片处理完成");
         }
     }