liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化历史消息处理

Browse Source
This commit is contained in:
gaoyutao
2026-02-26 17:59:05 +08:00
parent 9fdafe542d
commit 05c67476e6
1 changed files with 28 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java
+28 -74
View File
@@ -17,9 +17,9 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
@@ -37,6 +37,9 @@ public class TimeRangeReConsumer implements DisposableBean {
@Autowired(required = false)
private MessageHistoryDataHandler messageHandler;
// 从配置文件读取队列数量
@Value("${queues:4}")
private int totalQueues; // 测试环境配置为4，正式环境配置为16
// 时间范围配置
private static final String START_TIME = "2026-02-02 06:00:00";
@@ -184,16 +187,29 @@ public class TimeRangeReConsumer implements DisposableBean {
}
}
// 2. 超过结束时间，标记队列完成
// 2. 超过结束时间，标记队列完成
if (bornTimestamp > endTimestamp) {
log.info("队列 {} 已超过结束时间，标记为完成", queueId);
completedQueues.add(queueId);
boolean allQueuesNowCompleted = false;
// 检查是否所有队列都完成
if (completedQueues.size() >= 4) {
log.info("所有队列均已超过结束时间，第一阶段完成");
shouldContinue.set(false);
synchronized (completedQueues) {
boolean isNewCompletion = completedQueues.add(queueId);
if (isNewCompletion) {
log.info("队列 {} 已超过结束时间，标记为完成", queueId);
}
// 检查是否所有队列都完成 - 必须在同步块中检查
if (completedQueues.size() >= totalQueues && shouldContinue.get()) {
allQueuesNowCompleted = true;
shouldContinue.set(false);
}
}
// 在同步块外打印完成日志，确保只打印一次
if (allQueuesNowCompleted) {
log.info("所有队列均已超过结束时间，第一阶段完成");
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
@@ -226,9 +242,6 @@ public class TimeRangeReConsumer implements DisposableBean {
consumer.start();
log.info("时间范围重消费者启动成功，消费者组: {}", consumerGroup);
// 预热：查找大致的起始位置
findApproximateStartOffset(consumer, startTimestamp);
// 监控第一阶段完成
waitForPhase1Completion(consumer);
@@ -329,8 +342,8 @@ public class TimeRangeReConsumer implements DisposableBean {
// 每30秒打印一次进度
if (checkCount % 6 == 0) {
log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4",
currentMessageCount, totalCached, messageCache.size(), completedQueues.size());
log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/{}",
currentMessageCount, totalCached, messageCache.size(), completedQueues.size(), totalQueues);
}
// 检查是否有进度
@@ -342,7 +355,7 @@ public class TimeRangeReConsumer implements DisposableBean {
}
// 如果30秒内没有新消息，认为第一阶段完成
if (noProgressCount >= 6 && completedQueues.size() >= 4) {
if (noProgressCount >= 6 && completedQueues.size() >= totalQueues) {
log.info("30秒内无新消息且所有队列完成，第一阶段完成");
shouldContinue.set(false);
break;
@@ -542,65 +555,6 @@ public class TimeRangeReConsumer implements DisposableBean {
log.info("✓ 重消费者状态已重置");
}
/**
* 查找大致的起始offset - 修复版（处理异常）
*/
private void findApproximateStartOffset(DefaultMQPushConsumer consumer, long startTimestamp) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
log.info("查找起始offset，目标时间: {}", sdf.format(new Date(startTimestamp)));
try {
// 获取主题队列
Set<MessageQueue> queues = consumer.fetchSubscribeMessageQueues(MessageCodeEnum.TR_AGENT_UP.getCode());
if (queues == null || queues.isEmpty()) {
log.info("无队列信息，跳过offset查找");
return;
}
log.info("找到{}个队列，开始设置起始offset...", queues.size());
for (MessageQueue queue : queues) {
try {
// 搜索指定时间戳的offset
long offset = consumer.searchOffset(queue, startTimestamp);
if (offset >= 0) {
// 获取队列的offset范围
long minOffset = consumer.minOffset(queue);
long maxOffset = consumer.maxOffset(queue);
// 确保offset在有效范围内
if (offset < minOffset) offset = minOffset;
if (offset > maxOffset) offset = maxOffset;
queueCurrentOffsets.put(queue.getQueueId(), offset);
log.info("队列{}: offset={} (范围: {}-{})",
queue.getQueueId(), offset, minOffset, maxOffset);
} else {
// 使用最小offset - 使用新方法避免过时API
try {
long minOffset = consumer.minOffset(queue);
queueCurrentOffsets.put(queue.getQueueId(), minOffset);
log.info("队列{}: 使用最小offset={}", queue.getQueueId(), minOffset);
} catch (Exception e) {
// 如果minOffset也失败，使用0作为兜底
log.warn("队列{}获取最小offset失败，使用0作为兜底: {}", queue.getQueueId(), e.getMessage());
queueCurrentOffsets.put(queue.getQueueId(), 0L);
}
}
} catch (Exception e) {
log.warn("队列{}设置offset失败，使用0作为兜底: {}", queue.getQueueId(), e.getMessage());
queueCurrentOffsets.put(queue.getQueueId(), 0L);
}
}
log.info("起始offset设置完成");
} catch (Exception e) {
log.warn("查找起始offset失败，将使用默认时间戳方式: {}", e.getMessage());
}
}
/**
* 打印第一阶段进度
@@ -611,8 +565,8 @@ public class TimeRangeReConsumer implements DisposableBean {
totalCached += queue.size();
}
log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4",
receivedCount.get(), totalCached, messageCache.size(), completedQueues.size());
log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/{}",
receivedCount.get(), totalCached, messageCache.size(), completedQueues.size(), totalQueues);
}
/**