|
|
@@ -750,26 +750,29 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
}
|
|
|
|
|
|
public void sync(MidjourneyUserConversation conversation) throws IOException {
|
|
|
+ if (conversation.getFinishTime() != null) {
|
|
|
+ TASK_IMG_SIZE_MAP.remove(conversation.getTaskId());
|
|
|
+ }
|
|
|
if ((StringUtils.isNotBlank(conversation.getProgress()) && progress.contains(conversation.getProgress())) || status.contains(conversation.getStatus())) {
|
|
|
- log.info("同步任务:{},进度:{}",conversation.getTaskId(),conversation.getProgress());
|
|
|
- Long userId = conversation.getUserId();
|
|
|
- MidjourneyUserConversation dbConversation = conversationMapper.selectOne(Wrappers.lambdaQuery(MidjourneyUserConversation.class).eq(MidjourneyUserConversation::getTaskId, conversation.getTaskId())
|
|
|
- .eq(MidjourneyUserConversation::getUserId, userId).orderByDesc(MidjourneyUserConversation::getId).last(" limit 1"));
|
|
|
- if (dbConversation != null) {
|
|
|
- if ("SUCCESS".equals(dbConversation.getStatus()) || "FAILURE".equals(dbConversation.getStatus())) {
|
|
|
- return;
|
|
|
- }
|
|
|
- if (conversation.getProgressNum() <= dbConversation.getProgressNum() && !conversation.getStatus().equals("FAILURE")){
|
|
|
- return;
|
|
|
- }
|
|
|
- conversation.setOriginalImageUrl(conversation.getImageUrl());
|
|
|
- conversation.setAction(dbConversation.getAction());
|
|
|
- conversation.setMode(dbConversation.getMode());
|
|
|
- conversation.setId(dbConversation.getId());
|
|
|
- conversationMapper.updateById(conversation);
|
|
|
- //失败返还次数
|
|
|
- if ("FAILURE".equals(conversation.getStatus()) && !dbConversation.getStatus().equals("MODEL")){
|
|
|
- Integer mode = dbConversation.getMode();
|
|
|
+ log.info("同步任务:{},进度:{}",conversation.getTaskId(),conversation.getProgress());
|
|
|
+ Long userId = conversation.getUserId();
|
|
|
+ MidjourneyUserConversation dbConversation = conversationMapper.selectOne(Wrappers.lambdaQuery(MidjourneyUserConversation.class).eq(MidjourneyUserConversation::getTaskId, conversation.getTaskId())
|
|
|
+ .eq(MidjourneyUserConversation::getUserId, userId).orderByDesc(MidjourneyUserConversation::getId).last(" limit 1"));
|
|
|
+ if (dbConversation != null) {
|
|
|
+ if ("SUCCESS".equals(dbConversation.getStatus()) || "FAILURE".equals(dbConversation.getStatus())) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (conversation.getProgressNum() <= dbConversation.getProgressNum() && !conversation.getStatus().equals("FAILURE")){
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ conversation.setOriginalImageUrl(conversation.getImageUrl());
|
|
|
+ conversation.setAction(dbConversation.getAction());
|
|
|
+ conversation.setMode(dbConversation.getMode());
|
|
|
+ conversation.setId(dbConversation.getId());
|
|
|
+ conversationMapper.updateById(conversation);
|
|
|
+ //失败返还次数
|
|
|
+ if ("FAILURE".equals(conversation.getStatus()) && !dbConversation.getStatus().equals("MODEL")){
|
|
|
+ Integer mode = dbConversation.getMode();
|
|
|
// if (StringUtils.equals("未知频道",conversation.getFailReason())) {
|
|
|
// redisService.hdel(mode == 1 ? RedisService.key.MIDJOURNEY_FAST_ACCOUNT.getName() : RedisService.key.MIDJOURNEY_ACCOUNT.getName(),dbConversation.getInstanceId().toString());
|
|
|
// String finalAccountWithMinUsage = dbConversation.getInstanceId().toString();
|
|
|
@@ -780,37 +783,37 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
// }
|
|
|
// });
|
|
|
// }
|
|
|
- Object num = redisService.get(RedisService.key.MIDJOURNEY_RELAX_LIMIT.getName() + userId);
|
|
|
- LambdaUpdateWrapper<MidjourneyUser> wrapper = Wrappers.lambdaUpdate(MidjourneyUser.class)
|
|
|
- .eq(MidjourneyUser::getId, userId);
|
|
|
-
|
|
|
- // 使用Redis分布式锁避免重复扣除次数
|
|
|
- String lockKey = "sync:lock:" + conversation.getTaskId() + ":" + userId;
|
|
|
- // 尝试加锁
|
|
|
- boolean isLocked = redisService.tryLock(lockKey, 5);
|
|
|
- if (!isLocked) {
|
|
|
- // 未获取锁,避免重复操作
|
|
|
- return;
|
|
|
+ Object num = redisService.get(RedisService.key.MIDJOURNEY_RELAX_LIMIT.getName() + userId);
|
|
|
+ LambdaUpdateWrapper<MidjourneyUser> wrapper = Wrappers.lambdaUpdate(MidjourneyUser.class)
|
|
|
+ .eq(MidjourneyUser::getId, userId);
|
|
|
+
|
|
|
+ // 使用Redis分布式锁避免重复扣除次数
|
|
|
+ String lockKey = "sync:lock:" + conversation.getTaskId() + ":" + userId;
|
|
|
+ // 尝试加锁
|
|
|
+ boolean isLocked = redisService.tryLock(lockKey, 5);
|
|
|
+ if (!isLocked) {
|
|
|
+ // 未获取锁,避免重复操作
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ if (mode == 1){
|
|
|
+ num = redisService.incr(RedisService.key.MIDJOURNEY_FAST_LIMIT.getName() + userId, 1L);
|
|
|
+ wrapper.set(MidjourneyUser::getMjFastNum, num);
|
|
|
+ midjourneyUserMapper.update(null, wrapper);
|
|
|
}
|
|
|
- try {
|
|
|
- if (mode == 1){
|
|
|
- num = redisService.incr(RedisService.key.MIDJOURNEY_FAST_LIMIT.getName() + userId, 1L);
|
|
|
- wrapper.set(MidjourneyUser::getMjFastNum, num);
|
|
|
+ if (mode == 2){
|
|
|
+ if (num != null) {
|
|
|
+ num = redisService.incr(RedisService.key.MIDJOURNEY_RELAX_LIMIT.getName() + userId, 1L);
|
|
|
+ wrapper.set(MidjourneyUser::getMjRelaxNum, num);
|
|
|
midjourneyUserMapper.update(null, wrapper);
|
|
|
}
|
|
|
- if (mode == 2){
|
|
|
- if (num != null) {
|
|
|
- num = redisService.incr(RedisService.key.MIDJOURNEY_RELAX_LIMIT.getName() + userId, 1L);
|
|
|
- wrapper.set(MidjourneyUser::getMjRelaxNum, num);
|
|
|
- midjourneyUserMapper.update(null, wrapper);
|
|
|
- }
|
|
|
- }
|
|
|
- } finally {
|
|
|
- // 释放锁
|
|
|
- redisService.unlock(lockKey);
|
|
|
}
|
|
|
+ } finally {
|
|
|
+ // 释放锁
|
|
|
+ redisService.unlock(lockKey);
|
|
|
}
|
|
|
}
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -1097,10 +1100,14 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
});
|
|
|
|
|
|
TASK_EXECUTOR.execute(()->{
|
|
|
- handleCallback(conversation);
|
|
|
- conversationMapper.update(null, QueryWrapperUtils.buildUpdateWrapper((wrapper)->{
|
|
|
- wrapper.eq(MidjourneyUserConversation::getAction,"MODEL").eq(MidjourneyUserConversation::getTaskId, conversation.getTaskId()).set(MidjourneyUserConversation::getAction, conversation.getAction());
|
|
|
- }));
|
|
|
+ try {
|
|
|
+ handleCallback(conversation);
|
|
|
+ conversationMapper.update(null, QueryWrapperUtils.buildUpdateWrapper((wrapper)->{
|
|
|
+ wrapper.eq(MidjourneyUserConversation::getAction,"MODEL").eq(MidjourneyUserConversation::getTaskId, conversation.getTaskId()).set(MidjourneyUserConversation::getAction, conversation.getAction());
|
|
|
+ }));
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("发送任务出错 taskId:{}, error:{}", conversation.getTaskId(), StringUtil.getErrorText(e));
|
|
|
+ }
|
|
|
});
|
|
|
|
|
|
}
|
|
|
@@ -1124,6 +1131,7 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
ImgSize imgSize = new ImgSize();
|
|
|
imgSize.setHeight(image.getHeight());
|
|
|
imgSize.setWidth(image.getWidth());
|
|
|
+ image = null;
|
|
|
return imgSize;
|
|
|
}
|
|
|
|
|
|
@@ -1296,9 +1304,9 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
int iterationCount = 0;
|
|
|
int iterationRetries = 3;
|
|
|
// 获取当前时间
|
|
|
- //long currentTimeMillis = System.currentTimeMillis();
|
|
|
+ long currentTimeMillis = System.currentTimeMillis();
|
|
|
// 获取当前分钟,并计算下次任务执行的时间(下一个 5 分钟的 30 秒)
|
|
|
- //long nextTaskTime = (currentTimeMillis / (5 * 60 * 1000)) * (5 * 60 * 1000) + (5 * 60 * 1000) + 30 * 1000;
|
|
|
+ long nextTaskTime = (currentTimeMillis / (5 * 60 * 1000)) * (5 * 60 * 1000) + (5 * 60 * 1000) + 30 * 1000;
|
|
|
while (iterationCount < iterationRetries) {
|
|
|
iterationCount++;
|
|
|
// 查询需要上传的图片对话
|
|
|
@@ -1314,32 +1322,30 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
log.info("所有图片已处理完毕,结束同步任务");
|
|
|
break;
|
|
|
}
|
|
|
- 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);
|
|
|
- }
|
|
|
+ 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()));
|
|
|
}
|
|
|
- });
|
|
|
+ conversationMapper.update(null, Wrappers.lambdaUpdate(MidjourneyUserConversation.class)
|
|
|
+ .set(StringUtils.isNotBlank(newImageUrl), MidjourneyUserConversation::getImageUrl, newImageUrl)
|
|
|
+ .set(MidjourneyUserConversation::getImageInvalid, !uploadSuccess)
|
|
|
+ .eq(MidjourneyUserConversation::getId, conversation.getId()));
|
|
|
+ }
|
|
|
// 防止高频次查询,增加短暂的休眠时间
|
|
|
try {
|
|
|
int sleepTime = imgConversations.size() < batchSize ? 10000 : 1000;
|
|
|
@@ -1349,10 +1355,10 @@ public class MidjourneyServiceImpl implements MidjourneyService {
|
|
|
Thread.currentThread().interrupt();
|
|
|
}
|
|
|
// 判断当前时间是否接近下次任务执行时间
|
|
|
- //if (System.currentTimeMillis() > nextTaskTime - 30000) {
|
|
|
- // log.info("接近下次任务执行时间,提前结束同步任务");
|
|
|
- // break;
|
|
|
- //}
|
|
|
+ if (System.currentTimeMillis() > nextTaskTime - 30000) {
|
|
|
+ log.info("接近下次任务执行时间,提前结束同步任务");
|
|
|
+ break;
|
|
|
+ }
|
|
|
// 增加页码,查询下一页数据
|
|
|
pageNum++;
|
|
|
}
|