luoweijian
/
ai-evaluator
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

bug修复

master
luogw 2 months ago
parent
b481f23cad
commit
77efd0b2c7
3 changed files with 196 additions and 118 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 19
      src/main/java/com/project/interaction/application/impl/GenerateQuestionQueueSchedulerService.java
  2. 293
      src/main/java/com/project/interaction/domain/service/impl/GenerateQuestionQueueServiceImpl.java
  3. 2
      src/main/resources/application-dev.yml

19
src/main/java/com/project/interaction/application/impl/GenerateQuestionQueueSchedulerService.java
View File

@ -19,9 +19,9 @@ public class GenerateQuestionQueueSchedulerService {
/** /**
* 定时处理队列中的待重试项 * 定时处理队列中的待重试项
* 每分钟执行一次 * 上一次执行完成后间隔 60s 再次执行，避免重试耗时超过周期时背靠背/重入
*/ */
@Scheduled(fixedRate = 60000) @Scheduled(fixedDelay = 60000)
public void processQueuedItems() { public void processQueuedItems() {
try { try {
log.debug(">>> [定时任务] 开始处理队列中的待重试题目生成请求..."); log.debug(">>> [定时任务] 开始处理队列中的待重试题目生成请求...");
@ -30,20 +30,5 @@ public class GenerateQuestionQueueSchedulerService {
log.error(">>> [定时任务] 处理队列时发生异常, 错误: {}", e.getMessage(), e); log.error(">>> [定时任务] 处理队列时发生异常, 错误: {}", e.getMessage(), e);
} }
} }
/**
* 定时监控队列状态
* 每5分钟执行一次，监控队列大小和消费状态
*/
@Scheduled(fixedRate = 300000) // 每300秒执行一次（300000毫秒）
public void monitorQueueStatus() {
try {
log.debug(">>> [定时任务] 开始监控队列状态...");
// 这里可以添加队列监控逻辑，如记录队列大小、消费暂停/恢复状态等
log.debug(">>> [定时任务] 队列监控完成");
} catch (Exception e) {
log.error(">>> [定时任务] 监控队列时发生异常, 错误: {}", e.getMessage(), e);
}
}
} }

293
src/main/java/com/project/interaction/domain/service/impl/GenerateQuestionQueueServiceImpl.java
View File

@ -1,133 +1,194 @@
package com.project.interaction.domain.service.impl; package com.project.interaction.domain.service.impl;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.project.interaction.domain.dto.GenerateQuestionQueueDTO; import com.project.interaction.domain.dto.GenerateQuestionQueueDTO;
import com.project.interaction.domain.service.GenerateQuestionQueueService; import com.project.interaction.domain.service.GenerateQuestionQueueService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMap;
import org.redisson.api.RScoredSortedSet;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.*; import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap; import java.util.Collection;
import java.util.concurrent.PriorityBlockingQueue; import java.util.Comparator;
import java.util.concurrent.atomic.AtomicInteger; import java.util.List;
import java.util.concurrent.locks.ReentrantLock; import java.util.Map;
import java.util.UUID;
/**
* 生成题目重试队列 - 基于 Redisson 的持久化实现
* <p>
* 数据结构：
* <ul>
* <li>{@code items} (RMap) itemId -> DTO(JSON)，队列内容的唯一真相来源</li>
* <li>{@code schedule} (RScoredSortedSet) itemId，score = 下次可重试时间(epoch ms)，用于延迟/退避</li>
* </ul>
* 相比原内存实现，重启不丢失、支持指数退避；当前不限制重试次数，失败项将持续按退避重试。
*/
@Service @Service
@Slf4j @Slf4j
public class GenerateQuestionQueueServiceImpl implements GenerateQuestionQueueService { public class GenerateQuestionQueueServiceImpl implements GenerateQuestionQueueService {
/** 重试基础间隔（秒），退避从此值开始按 2 的幂增长 */
@Value("${question.queue.retry-interval:60}") @Value("${question.queue.retry-interval:60}")
private Long retryInterval; private long retryIntervalSeconds;
private PriorityBlockingQueue<GenerateQuestionQueueDTO> queue; /** 退避间隔上限（秒） */
@Value("${question.queue.max-retry-interval:1800}")
/*按照权限降序、时间升序*/ private long maxRetryIntervalSeconds;
@PostConstruct
public void init() { private static final String ITEMS_KEY = "question:gen:queue:items";
this.queue = new PriorityBlockingQueue<>( private static final String SCHEDULE_KEY = "question:gen:queue:schedule";
10000,
Comparator.comparing(GenerateQuestionQueueDTO::getWeight, Comparator.reverseOrder())
.thenComparing(GenerateQuestionQueueDTO::getAddTime)
.thenComparing(GenerateQuestionQueueDTO::getItemId)
);
}
private final Map<String, GenerateQuestionQueueDTO> itemMap = new ConcurrentHashMap<>(); /** 自持 ObjectMapper：注册 JavaTime，日期写为 ISO 文本，忽略未知字段以保证前后兼容 */
private static final ObjectMapper MAPPER = JsonMapper.builder()
.addModule(new JavaTimeModule())
.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.build();
private final AtomicInteger currentSize = new AtomicInteger(0); @Autowired
private RedissonClient redissonClient;
private final ReentrantLock capacityLock = new ReentrantLock(); private RMap<String, String> items() {
return redissonClient.getMap(ITEMS_KEY, StringCodec.INSTANCE);
}
private RScoredSortedSet<String> schedule() {
return redissonClient.getScoredSortedSet(SCHEDULE_KEY, StringCodec.INSTANCE);
}
@Override @Override
public boolean addItem(GenerateQuestionQueueDTO item) { public boolean addItem(GenerateQuestionQueueDTO item) {
capacityLock.lock(); if (item == null) {
try { return false;
if (item.getItemId() == null) { }
item.setItemId(UUID.randomUUID().toString()); if (item.getItemId() == null) {
} item.setItemId(UUID.randomUUID().toString());
if (item.getAddTime() == null) { }
item.setAddTime(LocalDateTime.now()); if (item.getAddTime() == null) {
} item.setAddTime(LocalDateTime.now());
if (item.getRetryCount() == null) { }
item.setRetryCount(0); if (item.getRetryCount() == null) {
} item.setRetryCount(0);
}
itemMap.put(item.getItemId(), item); String json = serialize(item);
queue.offer(item); if (json == null) {
currentSize.incrementAndGet(); return false;
}
log.info(">>> [队列管理] 成功添加队列项, ItemId: {}, 当前队列大小: {}, 添加时间: {}", RMap<String, String> items = items();
item.getItemId(), currentSize.get(), item.getAddTime()); // 先写内容，再登记调度：若中途异常，getRetryItems 会自愈清理孤立调度项
items.put(item.getItemId(), json);
schedule().add(System.currentTimeMillis(), item.getItemId());
return true; log.info(">>> [队列管理] 成功添加队列项, ItemId: {}, 当前队列大小: {}, 添加时间: {}",
} finally { item.getItemId(), items.size(), item.getAddTime());
capacityLock.unlock(); return true;
}
} }
@Override @Override
public List<GenerateQuestionQueueDTO> getRetryItems() { public List<GenerateQuestionQueueDTO> getRetryItems() {
capacityLock.lock(); long now = System.currentTimeMillis();
try { RMap<String, String> items = items();
List<GenerateQuestionQueueDTO> retryItems = new ArrayList<>(); RScoredSortedSet<String> schedule = schedule();
while (!queue.isEmpty()) { // 取出所有到期项（score <= now）。此处不移除，保证进程崩溃时重试项不丢失
GenerateQuestionQueueDTO item = queue.poll(); Collection<String> dueIds = schedule.valueRange(0, true, (double) now, true);
if (item == null) { if (dueIds == null || dueIds.isEmpty()) {
break; log.debug(">>> [队列管理] 当前没有到期待重试的项");
} return new ArrayList<>();
}
GenerateQuestionQueueDTO latestItem = itemMap.get(item.getItemId());
if (latestItem == null) {
// queue 中残留项（已被 removeItem/removeByTaskId 移除），直接丢弃
continue;
}
retryItems.add(latestItem);
}
return retryItems; List<GenerateQuestionQueueDTO> retryItems = new ArrayList<>();
} finally { for (String itemId : dueIds) {
capacityLock.unlock(); String json = items.get(itemId);
if (json == null) {
// 内容已被删除，清理残留调度项
schedule.remove(itemId);
continue;
}
GenerateQuestionQueueDTO item = deserialize(json);
if (item == null) {
// 反序列化失败（脏数据），丢弃避免卡死
items.remove(itemId);
schedule.remove(itemId);
continue;
}
retryItems.add(item);
} }
// 保持原有优先级：权重降序、添加时间升序、itemId 升序
retryItems.sort(Comparator
.comparing(GenerateQuestionQueueDTO::getWeight,
Comparator.nullsLast(Comparator.reverseOrder()))
.thenComparing(GenerateQuestionQueueDTO::getAddTime,
Comparator.nullsLast(Comparator.naturalOrder()))
.thenComparing(GenerateQuestionQueueDTO::getItemId,
Comparator.nullsLast(Comparator.naturalOrder())));
log.info(">>> [队列管理] 获取到 {} 个待重试的项", retryItems.size());
return retryItems;
} }
@Override @Override
public boolean removeItem(String itemId) { public boolean removeItem(String itemId) {
capacityLock.lock(); if (itemId == null) {
try {
GenerateQuestionQueueDTO removed = itemMap.remove(itemId);
if (removed != null) {
currentSize.decrementAndGet();
log.info(">>> [队列管理] 成功移除队列项, ItemId: {}, 当前队列大小: {}", itemId, currentSize.get());
return true;
}
return false; return false;
} finally {
capacityLock.unlock();
} }
boolean existed = items().remove(itemId) != null;
schedule().remove(itemId);
if (existed) {
log.info(">>> [队列管理] 成功移除队列项, ItemId: {}, 当前队列大小: {}", itemId, items().size());
}
return existed;
} }
@Override @Override
public int getQueueSize() { public int getQueueSize() {
return currentSize.get(); return items().size();
} }
@Override
public void requeue(GenerateQuestionQueueDTO item) { public void requeue(GenerateQuestionQueueDTO item) {
if (item == null || item.getItemId() == null) { if (item == null || item.getItemId() == null) {
return; return;
} }
capacityLock.lock(); String itemId = item.getItemId();
try { RMap<String, String> items = items();
if (!itemMap.containsKey(item.getItemId())) {
currentSize.incrementAndGet(); // 仅当该项仍在追踪中时才重新入队；若已被 removeItem/removeByTaskId 删除，则不再复活
} if (!items.containsKey(itemId)) {
itemMap.put(item.getItemId(), item); log.info(">>> [队列管理] 项已被删除, 跳过重新入队, ItemId: {}", itemId);
queue.offer(item); return;
} finally { }
capacityLock.unlock();
int retryCount = item.getRetryCount() == null ? 0 : item.getRetryCount();
// 指数退避：下次重试时间 = now + min(base * 2^(retryCount-1), cap)；当前不限制重试次数
long delaySeconds = computeBackoffSeconds(retryCount);
long nextRetryMillis = System.currentTimeMillis() + delaySeconds * 1000L;
String json = serialize(item);
if (json == null) {
return;
} }
items.put(itemId, json);
schedule().add((double) nextRetryMillis, itemId);
log.info(">>> [队列管理] 重新入队等待重试, ItemId: {}, 重试次数: {}, {}秒后重试",
itemId, retryCount, delaySeconds);
} }
@Override @Override
@ -135,23 +196,55 @@ public class GenerateQuestionQueueServiceImpl implements GenerateQuestionQueueSe
if (taskId == null) { if (taskId == null) {
return; return;
} }
capacityLock.lock(); RMap<String, String> items = items();
try { RScoredSortedSet<String> schedule = schedule();
List<String> keysToRemove = new ArrayList<>();
for (Map.Entry<String, GenerateQuestionQueueDTO> entry : itemMap.entrySet()) { List<String> keysToRemove = new ArrayList<>();
if (taskId.equals(entry.getValue().getTaskId())) { for (Map.Entry<String, String> entry : items.readAllEntrySet()) {
keysToRemove.add(entry.getKey()); if (taskId.equals(parseTaskId(entry.getValue()))) {
} keysToRemove.add(entry.getKey());
}
for (String key : keysToRemove) {
itemMap.remove(key);
currentSize.decrementAndGet();
}
if (!keysToRemove.isEmpty()) {
log.info(">>> [队列管理] 根据TaskId移除队列项, TaskId: {}, 移除数量: {}", taskId, keysToRemove.size());
} }
} finally { }
capacityLock.unlock(); for (String key : keysToRemove) {
items.remove(key);
schedule.remove(key);
}
if (!keysToRemove.isEmpty()) {
log.info(">>> [队列管理] 根据TaskId移除队列项, TaskId: {}, 移除数量: {}", taskId, keysToRemove.size());
}
}
/** 计算退避秒数：base * 2^(retryCount-1)，上限 maxRetryIntervalSeconds */
private long computeBackoffSeconds(int retryCount) {
int exp = Math.min(Math.max(0, retryCount - 1), 20);
long delay = (long) (retryIntervalSeconds * Math.pow(2, exp));
return Math.min(delay, maxRetryIntervalSeconds);
}
private String serialize(GenerateQuestionQueueDTO item) {
try {
return MAPPER.writeValueAsString(item);
} catch (Exception e) {
log.error(">>> [队列管理] 序列化队列项失败, ItemId: {}, 原因: {}", item.getItemId(), e.getMessage(), e);
return null;
}
}
private GenerateQuestionQueueDTO deserialize(String json) {
try {
return MAPPER.readValue(json, GenerateQuestionQueueDTO.class);
} catch (Exception e) {
log.error(">>> [队列管理] 反序列化队列项失败, 原因: {}, 内容: {}", e.getMessage(), json);
return null;
}
}
private Long parseTaskId(String json) {
try {
JsonNode node = MAPPER.readTree(json).get("taskId");
return node == null || node.isNull() ? null : node.asLong();
} catch (Exception e) {
return null;
} }
} }
} }

2
src/main/resources/application-dev.yml
View File

@ -95,7 +95,7 @@ question:
generation: generation:
# 限流速率：每秒允许的API请求数 # 限流速率：每秒允许的API请求数
rate-limit: 20 rate-limit: 20
downgrade: true downgrade: false
# 队列配置 # 队列配置
queue: queue:
# 重试间隔（秒） # 重试间隔（秒）

Write Preview
Loading…
Cancel
Save