|
|
@@ -0,0 +1,1825 @@
|
|
|
+package com.fs.system.service;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.common.RedisKeyCommon;
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
+import com.fasterxml.jackson.databind.node.ObjectNode;
|
|
|
+import com.fs.common.core.redis.RedisCache;
|
|
|
+import com.fs.course.domain.FsCourseWatchLog;
|
|
|
+import com.fs.course.mapper.FsCourseWatchLogMapper;
|
|
|
+import com.fs.sop.domain.QwSopLogs;
|
|
|
+import com.fs.sop.mapper.QwSopLogsMapper;
|
|
|
+import com.fs.system.domain.vo.WatchLogKeyVo;
|
|
|
+import com.fs.system.domain.vo.WatchLogScanBounds;
|
|
|
+import com.fs.system.support.MiniAppReplaceSupport;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.util.CollectionUtils;
|
|
|
+
|
|
|
+import javax.annotation.PreDestroy;
|
|
|
+import java.time.Instant;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.util.*;
|
|
|
+import java.util.concurrent.*;
|
|
|
+import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
+import java.util.concurrent.atomic.AtomicLong;
|
|
|
+import java.util.concurrent.atomic.LongAdder;
|
|
|
+import java.util.concurrent.locks.ReentrantLock;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 小程序 AppId 更换服务（支持同一天内多个旧 AppId 依次替换为同一个新 AppId）。
|
|
|
+ *
|
|
|
+ * <h3>整体流程</h3>
|
|
|
+ * <pre>
|
|
|
+ * 看课表(fs_course_watch_log) 按 log_id 游标扫描（按实际行数切少量并行分区，非 ID 步长）
|
|
|
+ * → 提取关联键(qwUserId,sopId,userId,externalId)
|
|
|
+ * → 查 SOP 表(qw_sop_logs) 匹配待发记录
|
|
|
+ * → 过滤 send_mini_app_id 属于「旧 AppId 集合」且 content_json 含旧值的记录
|
|
|
+ * → 批量 UPDATE send_mini_app_id 与 content_json 中的 miniprogramAppid
|
|
|
+ * </pre>
|
|
|
+ *
|
|
|
+ * <h3>Redis 幂等（key = fs_qw_miniapp_replace:{startTime}）</h3>
|
|
|
+ * 任务开始时若集合不含当前 oldAppId → 立刻写入 Redis（中断后不丢失登记）；
|
|
|
+ * 已含则跳过，避免同一天重复处理同一旧 AppId。
|
|
|
+ * 每次任务看课扫一遍；当前 oldAppId 优先替换为本次 newAppId，历史 oldAppId 并行替换为同一 newAppId。
|
|
|
+ *
|
|
|
+ * <h3>任务抢占</h3>
|
|
|
+ * 新任务到达 → taskEpoch 递增 + cancelled + 取消分区 Future；
|
|
|
+ * 旧任务在各检查点发现 epoch 不匹配后退出并释放 taskLock，新任务再执行。
|
|
|
+ *
|
|
|
+ * <h3>线程模型</h3>
|
|
|
+ * coordinatorExecutor（单线程 + 队列1）：串行跑主任务，DiscardOldest 保留最新排队任务；
|
|
|
+ * partitionExecutor（16~32 线程）：并行扫看课数据段；
|
|
|
+ * sopExecutor（16~40 线程，总 worker≤56）：查 SOP + 批量 UPDATE（看课扫完后再按轮次执行）；
|
|
|
+ * heartbeatExecutor（单线程定时）：每 5 分钟播报一次进度。
|
|
|
+ *
|
|
|
+ * <h3>分段策略</h3>
|
|
|
+ * 多段时一律按真实行数均分（ROW_NUMBER），每段约 12 万行，避免 log_id 空洞产生空段或拖尾段；
|
|
|
+ * 大数据量均分失败时降级单段，不回退 log_id 等距。
|
|
|
+ *
|
|
|
+ * <h3>CPU / 内存平滑</h3>
|
|
|
+ * 段任务滑动窗口 + 启动轻量错峰；SOP 背压 Semaphore；全局键去重有软上限；content_json 长度硬限制。
|
|
|
+ *
|
|
|
+ * <h3>安全保障</h3>
|
|
|
+ * 游标前进检测 + 循环硬上限防死循环；Semaphore 背压防拖垮 DB；总 worker ≤ 56；epoch 抢占可中断；
|
|
|
+ * 单批失败可丢弃并记日志，Future/分区等待有超时上限，避免无限阻塞导致任务卡死。
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Service
|
|
|
+public class ReplaceMiniAppHandleService {
|
|
|
+
|
|
|
+ // ======================== 批处理参数（安全硬上限 + 性能调优） ========================
|
|
|
+
|
|
|
+ private static final DateTimeFormatter DATETIME_FMT =
|
|
|
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault());
|
|
|
+
|
|
|
+ private static final int CPU_CORES = Math.max(4, Runtime.getRuntime().availableProcessors());
|
|
|
+ /** 看课扫描并行度（128 核约 32） */
|
|
|
+ private static final int PARALLELISM;
|
|
|
+ /** SOP 查改并行度（128 核约 24，与扫描合计不超过 MAX_TOTAL_WORKERS） */
|
|
|
+ private static final int SOP_PARALLELISM;
|
|
|
+ /** 扫描 + SOP 总 worker 硬上限，防止 16GB 机器线程/连接过多 */
|
|
|
+ private static final int MAX_TOTAL_WORKERS = 56;
|
|
|
+ /** 同时在途 SOP 任务上限（= SOP 线程 × 3，超出则扫描线程短暂等待，背压保护 DB） */
|
|
|
+ private static final int MAX_SOP_IN_FLIGHT;
|
|
|
+
|
|
|
+ static {
|
|
|
+ int scan = Math.min(32, Math.max(16, CPU_CORES / 4));
|
|
|
+ int sop = Math.min(40, Math.max(20, CPU_CORES / 3));
|
|
|
+ if (scan + sop > MAX_TOTAL_WORKERS) {
|
|
|
+ sop = Math.max(16, MAX_TOTAL_WORKERS - scan);
|
|
|
+ }
|
|
|
+ PARALLELISM = scan;
|
|
|
+ SOP_PARALLELISM = sop;
|
|
|
+ MAX_SOP_IN_FLIGHT = SOP_PARALLELISM * 3;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 每个并行段目标行数（按行均分，保证各段真实数据量接近） */
|
|
|
+ private static final long ROWS_PER_PARTITION = 120_000L;
|
|
|
+ private static final int MAX_PARTITIONS = 64;
|
|
|
+ private static final long SINGLE_PARTITION_FALLBACK_THRESHOLD = 100_000L;
|
|
|
+ /** 同时在途段数上限（渐进加大，见 resolveMaxInFlight） */
|
|
|
+ private static final int PARTITION_SUBMIT_WINDOW = Math.min(12, PARALLELISM + 2);
|
|
|
+ /** 启动阶段每提交 N 段停顿一次（毫秒），防止 CPU 瞬间打满 */
|
|
|
+ private static final long PARTITION_SUBMIT_STAGGER_MS = 300L;
|
|
|
+ private static final int PARTITION_SUBMIT_STAGGER_EVERY = 2;
|
|
|
+ /** 初始估算：每扫描线程约 2.5 万行/分钟（打 8 折留余量） */
|
|
|
+ private static final long EST_SCAN_ROWS_PER_THREAD_PER_MIN = 25_000L;
|
|
|
+ /** 单次从看课表拉取行数 */
|
|
|
+ private static final int CURSOR_BATCH_SIZE = 80_000;
|
|
|
+ /** 单次批量查 SOP 的关联键数量（复合键 IN 批次，略小有利于索引） */
|
|
|
+ private static final int SOP_KEY_BATCH_SIZE = 4000;
|
|
|
+ /** 历史 oldAppId 并行度（预估单日约 10 个旧 AppId，限制并行避免 CPU/DB 尖峰） */
|
|
|
+ private static final int HISTORICAL_PASS_PARALLELISM = Math.min(4, Math.max(2, CPU_CORES / 8));
|
|
|
+ /** 单次批量 UPDATE 条数 */
|
|
|
+ private static final int UPDATE_BATCH_SIZE = 2000;
|
|
|
+ /** 去重键达此数量即刷入全局 pending 集合，控制单段内存 */
|
|
|
+ private static final int FLUSH_KEY_THRESHOLD = 20_000;
|
|
|
+ /** 每扫 N 批看课也刷入 pending 集合 */
|
|
|
+ private static final int FLUSH_EVERY_CURSOR_BATCHES = 3;
|
|
|
+ /** 分区内游标循环硬上限（兜底） */
|
|
|
+ private static final int MAX_CURSOR_BATCHES_FALLBACK =
|
|
|
+ (int) Math.min(50_000, (ROWS_PER_PARTITION * MAX_PARTITIONS) / CURSOR_BATCH_SIZE + 500);
|
|
|
+ private static final int MAX_RETRY = 3;
|
|
|
+ private static final long INIT_RETRY_DELAY_MS = 100L;
|
|
|
+ private static final long TASK_WAIT_TIMEOUT_SECONDS = 300;
|
|
|
+ private static final int COORDINATOR_QUEUE_CAPACITY = 1;
|
|
|
+ private static final int PARTITION_QUEUE_CAPACITY = MAX_PARTITIONS;
|
|
|
+ private static final int SOP_QUEUE_CAPACITY = 384;
|
|
|
+ /** 等待单个 SOP 任务的最长时间（秒），超时后继续等并检查取消，避免假死无响应 */
|
|
|
+ private static final long SOP_FUTURE_WAIT_SECONDS = 60;
|
|
|
+ /** 单批 SOP Future 最多等待轮数（× SOP_FUTURE_WAIT_SECONDS 后取消并丢弃） */
|
|
|
+ private static final int SOP_FUTURE_MAX_TIMEOUT_ROUNDS = 120;
|
|
|
+ /** 扫描线程等 SOP 背压槽位的最长单次等待（秒），超时打日志 */
|
|
|
+ private static final long SOP_PERMIT_WAIT_SECONDS = 120L;
|
|
|
+ /** 等 SOP 槽位累计超过此秒数则丢弃本批，避免协调线程永久阻塞 */
|
|
|
+ private static final long SOP_PERMIT_MAX_WAIT_SECONDS = 600L;
|
|
|
+ /** 分区 CompletionService 连续空 poll 次数上限（×2秒），超时则取消在途段 */
|
|
|
+ private static final int PARTITION_POLL_IDLE_MAX = 150;
|
|
|
+ /** 历史轮次协调 Future 最长等待（秒） */
|
|
|
+ private static final long HISTORICAL_PASS_WAIT_SECONDS = 3600L;
|
|
|
+ /** Redis 集合过期时间（小时） */
|
|
|
+ private static final int REDIS_EXPIRE_HOURS = 24;
|
|
|
+ private static final String LOG_TAG = "[小程序换绑]";
|
|
|
+ /** 进度汇报间隔：5 分钟 */
|
|
|
+ private static final long PROGRESS_LOG_INTERVAL_MS = 300_000L;
|
|
|
+ /** content_json 最大处理长度，防止异常大 JSON 拖垮内存 */
|
|
|
+ private static final int MAX_CONTENT_JSON_LENGTH = 512_000;
|
|
|
+
|
|
|
+ private static final String PHASE_PREPARE = "启动准备";
|
|
|
+ private static final String PHASE_COUNT = "统计今日看课数量";
|
|
|
+ private static final String PHASE_SPLIT = "拆分并行任务";
|
|
|
+ private static final String PHASE_SCAN = "扫描看课表";
|
|
|
+ private static final String PHASE_SOP_PASS = "SOP逐轮查改";
|
|
|
+ private static final String PHASE_SOP_DRAIN = "SOP收尾";
|
|
|
+
|
|
|
+ private final ObjectMapper objectMapper = new ObjectMapper();
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private FsCourseWatchLogMapper watchLogMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private QwSopLogsMapper sopLogsMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private RedisCache redisCache;
|
|
|
+
|
|
|
+ // ======================== 专用线程池与并发控制 ========================
|
|
|
+
|
|
|
+ /** 单线程协调器：投递主任务、等待分区完成，不占用 Spring @Async 线程池 */
|
|
|
+ private final ThreadPoolExecutor coordinatorExecutor;
|
|
|
+ /** 分区 worker：只看课扫描 + 投递 SOP 任务 */
|
|
|
+ private final ThreadPoolExecutor partitionExecutor;
|
|
|
+ /** SOP worker：查 SOP + 批量 UPDATE */
|
|
|
+ private final ThreadPoolExecutor sopExecutor;
|
|
|
+ /** 历史 oldAppId 轮次协调（有限并行，不与当前轮同时跑） */
|
|
|
+ private final ExecutorService historicalPassExecutor;
|
|
|
+
|
|
|
+ /** 取消标志：新任务或 shutdown 时置 true，配合 epoch 让旧 worker 退出 */
|
|
|
+ private final AtomicBoolean cancelled = new AtomicBoolean(false);
|
|
|
+ /** 服务销毁标志：阻止新任务提交 */
|
|
|
+ private final AtomicBoolean destroyed = new AtomicBoolean(false);
|
|
|
+ /**
|
|
|
+ * 任务代际号：每次 replaceAppId / cancel / shutdown 时递增。
|
|
|
+ * 各 worker 持有提交时的 epoch，运行中若发现 epoch != taskEpoch 则视为已被取代。
|
|
|
+ */
|
|
|
+ private final AtomicLong taskEpoch = new AtomicLong(0);
|
|
|
+ /** 互斥锁：同一时刻只有一个 runReplaceTask 在执行（含等分区结束） */
|
|
|
+ private final ReentrantLock taskLock = new ReentrantLock();
|
|
|
+ /** 保护 activeFutures 的读写，避免 cancel 与 submit 竞态 */
|
|
|
+ private final Object futuresLock = new Object();
|
|
|
+ /** 当前任务已提交的分区 Future，供抢占时 cancel(true) 中断阻塞 IO */
|
|
|
+ private volatile List<Future<?>> activeFutures = Collections.emptyList();
|
|
|
+ /** 当前任务已提交的 SOP Future */
|
|
|
+ private volatile List<Future<long[]>> activeSopFutures = Collections.emptyList();
|
|
|
+ private final Object sopFuturesLock = new Object();
|
|
|
+
|
|
|
+ /** 全局统计（分区线程累加，LongAdder 无锁） */
|
|
|
+ private final LongAdder totalNeed = new LongAdder();
|
|
|
+ private final LongAdder totalUpdated = new LongAdder();
|
|
|
+ /** 诊断：各阶段累计计数，便于排查「有数据但结果为 0」 */
|
|
|
+ private final LongAdder statWatchLogRows = new LongAdder();
|
|
|
+ private final LongAdder statValidKeys = new LongAdder();
|
|
|
+ private final LongAdder statSopMatched = new LongAdder();
|
|
|
+ private final LongAdder statPartitionErrors = new LongAdder();
|
|
|
+ private final LongAdder statSopTasksInFlight = new LongAdder();
|
|
|
+ /** 因超时/异常主动丢弃的批次数（看课查库、SOP 查改、槽位等待等） */
|
|
|
+ private final LongAdder statBatchesDiscarded = new LongAdder();
|
|
|
+ /** 需改 SOP 的 id（任务级，用于最终报告列出漏改） */
|
|
|
+ private final Set<String> needUpdateSopIds = ConcurrentHashMap.newKeySet();
|
|
|
+ /** 已成功写入的 SOP id */
|
|
|
+ private final Set<String> successUpdateSopIds = ConcurrentHashMap.newKeySet();
|
|
|
+ /** 扫描阶段汇总的关联键（看课扫完后再按轮次查 SOP，扫描不被 SOP 背压阻塞） */
|
|
|
+ private final Set<WatchLogKeyVo> pendingSopKeys = ConcurrentHashMap.newKeySet();
|
|
|
+ /** 待处理关联键过多时只 warn 一次 */
|
|
|
+ private final AtomicBoolean pendingKeysWarned = new AtomicBoolean(false);
|
|
|
+ private static final int PENDING_KEYS_WARN_THRESHOLD = 1_500_000;
|
|
|
+ /** 最终报告最多打印的漏改 id 条数 */
|
|
|
+ private static final int MAX_MISSED_IDS_LOG = 2000;
|
|
|
+
|
|
|
+ /** 心跳汇报用：由协调线程写入，心跳线程只读 */
|
|
|
+ private volatile long progressEpoch = -1L;
|
|
|
+ private volatile long progressTaskStartMs = 0L;
|
|
|
+ private volatile String progressPhase = "";
|
|
|
+ private volatile long progressTotalRows = 0L;
|
|
|
+ private volatile int progressDonePartitions = 0;
|
|
|
+ private volatile int progressTotalPartitions = 0;
|
|
|
+ /** 当前任务每段允许的最大游标批次数 */
|
|
|
+ private volatile int taskMaxCursorBatches = MAX_CURSOR_BATCHES_FALLBACK;
|
|
|
+ /** 本任务 SOP 背压信号量（限制同时在途 SOP 任务数） */
|
|
|
+ private volatile Semaphore taskSopPermits;
|
|
|
+ /** 上次心跳时的看课扫描行数（用于算速度和检测卡住） */
|
|
|
+ private volatile long heartbeatLastScannedRows = 0L;
|
|
|
+ private volatile long heartbeatLastTickMs = 0L;
|
|
|
+ /** 根据当前速度推算的预计完成时刻（毫秒），每次进度汇报更新 */
|
|
|
+ private volatile long progressEstimatedFinishMs = 0L;
|
|
|
+ /** 当前 SOP 轮次描述（心跳展示） */
|
|
|
+ private volatile String progressSopPassLabel = "";
|
|
|
+ /** 本任务各 oldAppId 处理明细（协调线程写入，报告时读取） */
|
|
|
+ private final List<SopPassSummary> taskSopPassSummaries = new ArrayList<>();
|
|
|
+ /** 本任务 Redis 登记 oldAppId 总数 */
|
|
|
+ private volatile int taskRegistryOldAppCount = 0;
|
|
|
+
|
|
|
+ /** 进度心跳（与协调/分区线程隔离，避免长时间 DB 阻塞时无日志） */
|
|
|
+ private final ScheduledExecutorService heartbeatExecutor;
|
|
|
+ /** 防止重复心跳任务泄漏 */
|
|
|
+ private final Object heartbeatLock = new Object();
|
|
|
+ private volatile ScheduledFuture<?> activeHeartbeatFuture;
|
|
|
+
|
|
|
+ public ReplaceMiniAppHandleService() {
|
|
|
+ this.coordinatorExecutor = new ThreadPoolExecutor(
|
|
|
+ 1, 1, 0L, TimeUnit.MILLISECONDS,
|
|
|
+ new ArrayBlockingQueue<>(COORDINATOR_QUEUE_CAPACITY),
|
|
|
+ namedThreadFactory("miniapp-replace-coordinator"),
|
|
|
+ (r, e) -> {
|
|
|
+ if (!e.isShutdown()) {
|
|
|
+ log.info("{} 排队任务过多，丢弃最旧的一条，执行最新换绑请求", LOG_TAG);
|
|
|
+ e.getQueue().poll();
|
|
|
+ e.execute(r);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ this.partitionExecutor = new ThreadPoolExecutor(
|
|
|
+ PARALLELISM, PARALLELISM, 60L, TimeUnit.SECONDS,
|
|
|
+ new ArrayBlockingQueue<>(PARTITION_QUEUE_CAPACITY),
|
|
|
+ namedThreadFactory("miniapp-replace-partition"),
|
|
|
+ new ThreadPoolExecutor.CallerRunsPolicy());
|
|
|
+ this.partitionExecutor.allowCoreThreadTimeOut(true);
|
|
|
+ this.sopExecutor = new ThreadPoolExecutor(
|
|
|
+ SOP_PARALLELISM, SOP_PARALLELISM, 60L, TimeUnit.SECONDS,
|
|
|
+ new ArrayBlockingQueue<>(SOP_QUEUE_CAPACITY),
|
|
|
+ namedThreadFactory("miniapp-replace-sop"),
|
|
|
+ new ThreadPoolExecutor.CallerRunsPolicy());
|
|
|
+ this.sopExecutor.allowCoreThreadTimeOut(true);
|
|
|
+ this.historicalPassExecutor = Executors.newFixedThreadPool(
|
|
|
+ HISTORICAL_PASS_PARALLELISM,
|
|
|
+ namedThreadFactory("miniapp-replace-hist-pass"));
|
|
|
+ this.heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(
|
|
|
+ namedThreadFactory("miniapp-replace-heartbeat"));
|
|
|
+ }
|
|
|
+
|
|
|
+ // ======================== 对外入口 ========================
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 提交替换任务（异步）：HTTP 线程校验参数后立即返回，实际工作在协调线程执行。
|
|
|
+ * <p>注意：此处会先递增 epoch 并取消旧任务，再投递到 coordinatorExecutor。</p>
|
|
|
+ */
|
|
|
+ public void replaceAppId(String oldAppId, String newAppId, String startTime) {
|
|
|
+ String err = MiniAppReplaceSupport.validateReplaceParams(oldAppId, newAppId, startTime);
|
|
|
+ if (err != null) {
|
|
|
+ throw new IllegalArgumentException(err);
|
|
|
+ }
|
|
|
+ if (destroyed.get()) {
|
|
|
+ log.info("{} 服务已关闭，任务未提交", LOG_TAG);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ final long epoch = taskEpoch.incrementAndGet();
|
|
|
+ cancelled.set(true);
|
|
|
+ cancelActiveFutures();
|
|
|
+
|
|
|
+ try {
|
|
|
+ coordinatorExecutor.execute(() -> runReplaceTask(oldAppId, newAppId, startTime, epoch));
|
|
|
+ } catch (RejectedExecutionException e) {
|
|
|
+ log.error("{} 协调线程池拒绝任务 epoch={}", LOG_TAG, epoch, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 主任务体（仅在协调线程中运行）。
|
|
|
+ * <ol>
|
|
|
+ * <li>抢 taskLock，防止与上一任务重叠</li>
|
|
|
+ * <li>登记 Redis（oldAppId 立刻写入）</li>
|
|
|
+ * <li>分区并行扫描看课表 → 更新 SOP 表</li>
|
|
|
+ * <li>finally 释放锁，便于下一任务进入</li>
|
|
|
+ * </ol>
|
|
|
+ */
|
|
|
+ private void runReplaceTask(String oldAppId, String newAppId, String startTime, long epoch) {
|
|
|
+ if (epoch != taskEpoch.get()) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ boolean locked = false;
|
|
|
+ long startMs = System.currentTimeMillis();
|
|
|
+ ScheduledFuture<?> heartbeatFuture = null;
|
|
|
+ try {
|
|
|
+ locked = taskLock.tryLock(TASK_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
|
|
|
+ if (!locked) {
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, "已放弃", "上一个任务未在5分钟内结束");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (epoch != taskEpoch.get()) {
|
|
|
+ log.info("{} 任务已过期，跳过执行 | epoch={}", LOG_TAG, epoch);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ cancelled.set(false);
|
|
|
+ resetStats();
|
|
|
+
|
|
|
+ Set<String> registeredOldAppIds = registerOldAppIdAtStart(oldAppId, startTime);
|
|
|
+ if (registeredOldAppIds == null) {
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, "已结束", "今日已处理过或Redis写入失败");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ taskRegistryOldAppCount = registeredOldAppIds.size();
|
|
|
+ List<SopPass> sopPasses = buildSopPasses(registeredOldAppIds, oldAppId, newAppId);
|
|
|
+
|
|
|
+ heartbeatFuture = startProgressHeartbeat(epoch, startMs);
|
|
|
+ processUpdatePhase(sopPasses, startTime, epoch, oldAppId, newAppId, startMs);
|
|
|
+
|
|
|
+ if (epoch == taskEpoch.get() && !cancelled.get()) {
|
|
|
+ String finalStatus = statBatchesDiscarded.sum() > 0 ? "完成(有批次丢弃)" : "完成";
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, finalStatus, null);
|
|
|
+ } else {
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, "已中断", "被更新的换绑请求取代");
|
|
|
+ }
|
|
|
+
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, "已中断", "线程被中断");
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 任务异常终止 | 旧={} 新={}", LOG_TAG, oldAppId, newAppId, e);
|
|
|
+ logEnd(oldAppId, newAppId, startTime, startMs, "失败",
|
|
|
+ e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName());
|
|
|
+ } finally {
|
|
|
+ stopProgressHeartbeat(heartbeatFuture);
|
|
|
+ if (locked) {
|
|
|
+ // 仅当前 epoch 仍有效时清除取消标志，避免覆盖新任务/ shutdown 设置的 cancelled
|
|
|
+ if (epoch == taskEpoch.get()) {
|
|
|
+ cancelled.set(false);
|
|
|
+ }
|
|
|
+ synchronized (futuresLock) {
|
|
|
+ activeFutures = Collections.emptyList();
|
|
|
+ }
|
|
|
+ synchronized (sopFuturesLock) {
|
|
|
+ activeSopFutures = Collections.emptyList();
|
|
|
+ }
|
|
|
+ taskLock.unlock();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void resetStats() {
|
|
|
+ totalNeed.reset();
|
|
|
+ totalUpdated.reset();
|
|
|
+ statWatchLogRows.reset();
|
|
|
+ statValidKeys.reset();
|
|
|
+ statSopMatched.reset();
|
|
|
+ statPartitionErrors.reset();
|
|
|
+ statSopTasksInFlight.reset();
|
|
|
+ statBatchesDiscarded.reset();
|
|
|
+ needUpdateSopIds.clear();
|
|
|
+ successUpdateSopIds.clear();
|
|
|
+ pendingSopKeys.clear();
|
|
|
+ pendingKeysWarned.set(false);
|
|
|
+ progressSopPassLabel = "";
|
|
|
+ heartbeatLastScannedRows = 0L;
|
|
|
+ heartbeatLastTickMs = 0L;
|
|
|
+ progressEstimatedFinishMs = 0L;
|
|
|
+ taskSopPermits = new Semaphore(MAX_SOP_IN_FLIGHT);
|
|
|
+ taskSopPassSummaries.clear();
|
|
|
+ taskRegistryOldAppCount = 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 外部主动取消：递增 epoch 并中断分区线程 */
|
|
|
+ public void cancel() {
|
|
|
+ taskEpoch.incrementAndGet();
|
|
|
+ cancelled.set(true);
|
|
|
+ cancelActiveFutures();
|
|
|
+ log.info("{} 收到取消指令", LOG_TAG);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 是否有任务正持有 taskLock（含等待分区结束阶段） */
|
|
|
+ public boolean isRunning() {
|
|
|
+ return taskLock.isLocked();
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 中断当前任务所有分区与 SOP Future */
|
|
|
+ private void cancelActiveFutures() {
|
|
|
+ List<Future<?>> futures;
|
|
|
+ synchronized (futuresLock) {
|
|
|
+ futures = new ArrayList<>(activeFutures);
|
|
|
+ }
|
|
|
+ for (Future<?> future : futures) {
|
|
|
+ future.cancel(true);
|
|
|
+ }
|
|
|
+ List<Future<long[]>> sopFutures;
|
|
|
+ synchronized (sopFuturesLock) {
|
|
|
+ sopFutures = new ArrayList<>(activeSopFutures);
|
|
|
+ }
|
|
|
+ for (Future<long[]> future : sopFutures) {
|
|
|
+ future.cancel(true);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ======================== Redis：按 startTime 登记已处理 oldAppId ========================
|
|
|
+
|
|
|
+ private String redisKey(String startTime) {
|
|
|
+ return RedisKeyCommon.QW_MINAPP_REPLACE_KEY + startTime;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 读取当日已登记 oldAppId 集合（兼容旧版 Map / Set JSON）。
|
|
|
+ *
|
|
|
+ * @return 空集合表示 key 不存在；null 表示解析失败
|
|
|
+ */
|
|
|
+ private Set<String> loadRegisteredOldAppIds(String startTime) {
|
|
|
+ String key = redisKey(startTime);
|
|
|
+ Set<String> ids = new LinkedHashSet<>();
|
|
|
+ if (!redisCache.hasKey(key)) {
|
|
|
+ return ids;
|
|
|
+ }
|
|
|
+ String json = redisCache.getCacheObject(key);
|
|
|
+ if (StringUtils.isBlank(json)) {
|
|
|
+ return ids;
|
|
|
+ }
|
|
|
+ String trimmed = json.trim();
|
|
|
+ if (trimmed.startsWith("[")) {
|
|
|
+ try {
|
|
|
+ com.alibaba.fastjson.JSONArray arr = com.alibaba.fastjson.JSONArray.parseArray(trimmed);
|
|
|
+ if (arr != null) {
|
|
|
+ for (int i = 0; i < arr.size(); i++) {
|
|
|
+ String oldId = StringUtils.trimToNull(arr.getString(i));
|
|
|
+ if (oldId != null) {
|
|
|
+ ids.add(oldId);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return ids;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 解析Redis Set失败 key={}", LOG_TAG, key, e);
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ Map<String, ?> map = JSONObject.parseObject(trimmed, Map.class);
|
|
|
+ if (map != null) {
|
|
|
+ for (String oldId : map.keySet()) {
|
|
|
+ oldId = StringUtils.trimToNull(oldId);
|
|
|
+ if (oldId != null) {
|
|
|
+ ids.add(oldId);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return ids;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 解析Redis失败 key={}", LOG_TAG, key, e);
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return ids;
|
|
|
+ }
|
|
|
+
|
|
|
+ private boolean saveRegisteredOldAppIds(String startTime, Set<String> oldAppIds) {
|
|
|
+ try {
|
|
|
+ redisCache.setCacheObject(redisKey(startTime), JSONObject.toJSONString(oldAppIds),
|
|
|
+ REDIS_EXPIRE_HOURS, TimeUnit.HOURS);
|
|
|
+ return true;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 写入Redis失败 key={}", LOG_TAG, redisKey(startTime), e);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 登记当前 oldAppId；返回登记后全集。
|
|
|
+ *
|
|
|
+ * @return 全集；null = 已登记跳过 / Redis 异常
|
|
|
+ */
|
|
|
+ private Set<String> registerOldAppIdAtStart(String oldAppId, String startTime) {
|
|
|
+ Set<String> registered = loadRegisteredOldAppIds(startTime);
|
|
|
+ if (registered == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ if (registered.contains(oldAppId)) {
|
|
|
+ log.info("{} 今日已处理过该旧AppId，跳过 | oldAppId={} 日期={}", LOG_TAG, oldAppId, startTime);
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ registered.add(oldAppId);
|
|
|
+ if (!saveRegisteredOldAppIds(startTime, registered)) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return registered;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 构建 SOP 轮次：第 1 轮当前 oldAppId，其余历史 oldAppId 全部替换为本次 newAppId（每轮单 AppId 等号查库）。
|
|
|
+ */
|
|
|
+ private List<SopPass> buildSopPasses(Set<String> registeredOldAppIds, String currentOldAppId,
|
|
|
+ String currentNewAppId) {
|
|
|
+ List<SopPass> passes = new ArrayList<>();
|
|
|
+ passes.add(new SopPass(currentOldAppId, currentNewAppId, false));
|
|
|
+
|
|
|
+ List<String> historical = registeredOldAppIds.stream()
|
|
|
+ .filter(id -> !id.equals(currentOldAppId))
|
|
|
+ .sorted()
|
|
|
+ .collect(Collectors.toList());
|
|
|
+ for (String oldId : historical) {
|
|
|
+ passes.add(new SopPass(oldId, currentNewAppId, true));
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("{} SOP计划 | 共{}个旧AppId | 第1轮当前{}→{} | 历史{}轮均→{} | 历史完成后并行{}路",
|
|
|
+ LOG_TAG, registeredOldAppIds.size(), currentOldAppId, currentNewAppId,
|
|
|
+ historical.size(), currentNewAppId,
|
|
|
+ historical.isEmpty() ? 0 : Math.min(HISTORICAL_PASS_PARALLELISM, historical.size()));
|
|
|
+ synchronized (taskSopPassSummaries) {
|
|
|
+ taskSopPassSummaries.clear();
|
|
|
+ }
|
|
|
+ return passes;
|
|
|
+ }
|
|
|
+
|
|
|
+ // ======================== 分区调度：并行扫描看课表 ========================
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 看课并行扫描 → 汇总关联键 → SOP 当前 oldAppId 优先，再历史 oldAppId 并行。
|
|
|
+ */
|
|
|
+ private void processUpdatePhase(List<SopPass> sopPasses, String startTime, long epoch,
|
|
|
+ String currentOldAppId, String currentNewAppId, long taskStartMs)
|
|
|
+ throws InterruptedException {
|
|
|
+ checkRunning(epoch);
|
|
|
+ progressPhase = PHASE_COUNT;
|
|
|
+
|
|
|
+ WatchLogScanBounds bounds;
|
|
|
+ try {
|
|
|
+ bounds = watchLogMapper.selectWatchLogScanBounds(startTime);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 统计看课数据失败", LOG_TAG, e);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (bounds == null || bounds.getMinId() == null || bounds.getMaxId() == null
|
|
|
+ || bounds.getRowCount() == null || bounds.getRowCount() == 0) {
|
|
|
+ log.info("{} 今日无看课记录，无需处理 | 日期={}", LOG_TAG, startTime);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ long minId = bounds.getMinId();
|
|
|
+ long maxId = bounds.getMaxId();
|
|
|
+ long rowCount = bounds.getRowCount();
|
|
|
+ progressTotalRows = rowCount;
|
|
|
+ progressPhase = PHASE_SPLIT;
|
|
|
+
|
|
|
+ // 先打开始日志，再切分；大数据量用 ROW_NUMBER 按行均分，避免 log_id 空洞拖尾
|
|
|
+ progressEstimatedFinishMs = estimateInitialFinishMs(taskStartMs, rowCount);
|
|
|
+ log.info("{} 开始处理 | {} → {} | 日期={} | 看课共{}条 | 开始{} | 预计完成{}",
|
|
|
+ LOG_TAG, currentOldAppId, currentNewAppId, startTime, rowCount,
|
|
|
+ formatDateTime(taskStartMs), formatDateTime(progressEstimatedFinishMs));
|
|
|
+
|
|
|
+ List<Partition> partitions = buildPartitions(minId, maxId, rowCount, startTime);
|
|
|
+ int totalPartitions = partitions.size();
|
|
|
+ progressTotalPartitions = totalPartitions;
|
|
|
+ progressDonePartitions = 0;
|
|
|
+ long rowsPerPartition = (rowCount + totalPartitions - 1) / totalPartitions;
|
|
|
+ taskMaxCursorBatches = (int) Math.min(MAX_CURSOR_BATCHES_FALLBACK,
|
|
|
+ (rowsPerPartition + CURSOR_BATCH_SIZE - 1) / CURSOR_BATCH_SIZE + 200);
|
|
|
+ progressPhase = PHASE_SCAN;
|
|
|
+ log.info("{} 并行方案 | 分{}段扫看课 | {}线程 | 扫完后SOP {}轮（当前优先）",
|
|
|
+ LOG_TAG, totalPartitions, PARALLELISM, sopPasses.size());
|
|
|
+
|
|
|
+ CountDownLatch latch = new CountDownLatch(totalPartitions);
|
|
|
+ List<Future<?>> futures = new CopyOnWriteArrayList<>();
|
|
|
+ synchronized (futuresLock) {
|
|
|
+ activeFutures = futures;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ runPartitionsWithWindow(partitions, latch, futures, startTime, epoch, totalPartitions, rowCount);
|
|
|
+ progressDonePartitions = totalPartitions - (int) latch.getCount();
|
|
|
+
|
|
|
+ if (pendingSopKeys.isEmpty()) {
|
|
|
+ log.info("{} 看课扫描完成，无有效关联键，跳过SOP", LOG_TAG);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ List<WatchLogKeyVo> allKeys = new ArrayList<>(pendingSopKeys);
|
|
|
+ pendingSopKeys.clear();
|
|
|
+ log.info("{} 看课扫描完成 | 关联键{}个 | 开始SOP（当前oldAppId优先）",
|
|
|
+ LOG_TAG, allKeys.size());
|
|
|
+
|
|
|
+ List<Future<long[]>> sopFutures = Collections.synchronizedList(new ArrayList<>());
|
|
|
+ synchronized (sopFuturesLock) {
|
|
|
+ activeSopFutures = sopFutures;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ processAllSopPasses(allKeys, sopPasses, startTime, epoch, sopFutures);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ throw e;
|
|
|
+ } catch (Exception e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} SOP阶段异常，已结束剩余轮次", LOG_TAG, e);
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ synchronized (futuresLock) {
|
|
|
+ activeFutures = Collections.emptyList();
|
|
|
+ }
|
|
|
+ synchronized (sopFuturesLock) {
|
|
|
+ activeSopFutures = Collections.emptyList();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (statPartitionErrors.sum() > 0) {
|
|
|
+ log.error("{} 处理结束，有{}段任务曾发生异常，请检查上方错误日志", LOG_TAG, statPartitionErrors.sum());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 滑动窗口提交并行段：同时在途不超过 PARTITION_SUBMIT_WINDOW，避免一次性堆满队列。
|
|
|
+ */
|
|
|
+ private void runPartitionsWithWindow(List<Partition> partitions, CountDownLatch latch,
|
|
|
+ List<Future<?>> futures,
|
|
|
+ String startTime, long epoch,
|
|
|
+ int totalPartitions, long rowCount) throws InterruptedException {
|
|
|
+ ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(partitionExecutor);
|
|
|
+ int nextIndex = 0;
|
|
|
+ int inFlight = 0;
|
|
|
+ int submittedCount = 0;
|
|
|
+ int idlePolls = 0;
|
|
|
+
|
|
|
+ while (nextIndex < partitions.size() || inFlight > 0) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ while (nextIndex < partitions.size() && inFlight < resolveMaxInFlight(submittedCount, totalPartitions, rowCount)) {
|
|
|
+ Partition partition = partitions.get(nextIndex++);
|
|
|
+ if (partition.startId > partition.endId) {
|
|
|
+ latch.countDown();
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ Future<Void> future = completionService.submit(() -> {
|
|
|
+ try {
|
|
|
+ runOnePartition(partition, startTime, epoch);
|
|
|
+ } finally {
|
|
|
+ latch.countDown();
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ });
|
|
|
+ futures.add(future);
|
|
|
+ inFlight++;
|
|
|
+ submittedCount++;
|
|
|
+ if (shouldStaggerPartitionSubmit(submittedCount, totalPartitions, rowCount)) {
|
|
|
+ long staggerMs = (totalPartitions >= 8 || rowCount >= SINGLE_PARTITION_FALLBACK_THRESHOLD)
|
|
|
+ ? PARTITION_SUBMIT_STAGGER_MS / 2 : PARTITION_SUBMIT_STAGGER_MS;
|
|
|
+ Thread.sleep(staggerMs);
|
|
|
+ checkRunning(epoch);
|
|
|
+ }
|
|
|
+ } catch (RejectedExecutionException e) {
|
|
|
+ latch.countDown();
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 段任务提交被拒绝，丢弃 | logId={}-{}", LOG_TAG, partition.startId, partition.endId, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (inFlight == 0) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ Future<Void> completed = completionService.poll(2, TimeUnit.SECONDS);
|
|
|
+ if (completed != null) {
|
|
|
+ idlePolls = 0;
|
|
|
+ try {
|
|
|
+ completed.get();
|
|
|
+ } catch (ExecutionException e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 某段任务执行失败", LOG_TAG, e.getCause());
|
|
|
+ } catch (CancellationException ignored) {
|
|
|
+ // 任务被新请求取消
|
|
|
+ }
|
|
|
+ inFlight--;
|
|
|
+ progressDonePartitions = totalPartitions - (int) latch.getCount();
|
|
|
+ } else if (inFlight > 0) {
|
|
|
+ idlePolls++;
|
|
|
+ if (idlePolls >= PARTITION_POLL_IDLE_MAX) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ statBatchesDiscarded.add(inFlight);
|
|
|
+ log.error("{} 分区任务{}秒无进展，取消在途{}段并继续",
|
|
|
+ LOG_TAG, idlePolls * 2L, inFlight);
|
|
|
+ for (Future<?> f : futures) {
|
|
|
+ f.cancel(true);
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void runOnePartition(Partition partition, String startTime, long epoch) {
|
|
|
+ try {
|
|
|
+ updatePartition(partition, startTime, epoch);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ } catch (Exception e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 某段任务异常 | logId={}-{}", LOG_TAG, partition.startId, partition.endId, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 单分区：游标扫描看课表 → 汇总关联键（本阶段不查 SOP，避免背压阻塞扫描）。
|
|
|
+ */
|
|
|
+ private void updatePartition(Partition partition, String startTime, long epoch)
|
|
|
+ throws InterruptedException {
|
|
|
+ long cursor = partition.startId - 1;
|
|
|
+ int batchCount = 0;
|
|
|
+ int batchesSinceFlush = 0;
|
|
|
+
|
|
|
+ Set<WatchLogKeyVo> dedupKeys = new HashSet<>(8192);
|
|
|
+
|
|
|
+ while (cursor < partition.endId) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ if (++batchCount > taskMaxCursorBatches) {
|
|
|
+ log.error("{} 本段扫看课次数超限，强制结束 | 段范围logId={}-{} | 已扫{}批",
|
|
|
+ LOG_TAG, partition.startId, partition.endId, batchCount - 1);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<FsCourseWatchLog> logs = fetchWatchLogsWithRetry(startTime, cursor, partition.endId);
|
|
|
+ if (logs == null) {
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 看课查询连续失败，本段提前结束 | logId={}-{} cursor={}",
|
|
|
+ LOG_TAG, partition.startId, partition.endId, cursor);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (CollectionUtils.isEmpty(logs)) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+
|
|
|
+ FsCourseWatchLog lastLog = logs.get(logs.size() - 1);
|
|
|
+ Long lastLogId = lastLog != null ? lastLog.getLogId() : null;
|
|
|
+ if (lastLogId == null || lastLogId <= cursor) {
|
|
|
+ log.info("{} 看课游标没往前走，本段提前结束 | logId={}-{} cursor={}",
|
|
|
+ LOG_TAG, partition.startId, partition.endId, cursor);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ cursor = lastLogId;
|
|
|
+ statWatchLogRows.add(logs.size());
|
|
|
+ batchesSinceFlush++;
|
|
|
+
|
|
|
+ for (FsCourseWatchLog watchLog : logs) {
|
|
|
+ WatchLogKeyVo key = buildKeyOrNull(watchLog);
|
|
|
+ if (key != null) {
|
|
|
+ dedupKeys.add(key);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ logs.clear();
|
|
|
+
|
|
|
+ if (dedupKeys.size() >= FLUSH_KEY_THRESHOLD
|
|
|
+ || batchesSinceFlush >= FLUSH_EVERY_CURSOR_BATCHES) {
|
|
|
+ accumulatePendingKeys(dedupKeys, epoch);
|
|
|
+ batchesSinceFlush = 0;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (!dedupKeys.isEmpty()) {
|
|
|
+ accumulatePendingKeys(dedupKeys, epoch);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 扫描阶段：关联键写入全局集合（跨段去重），扫完后再统一 SOP */
|
|
|
+ private void accumulatePendingKeys(Set<WatchLogKeyVo> dedupKeys, long epoch) throws InterruptedException {
|
|
|
+ checkRunning(epoch);
|
|
|
+ for (WatchLogKeyVo key : dedupKeys) {
|
|
|
+ if (pendingSopKeys.add(key)) {
|
|
|
+ statValidKeys.increment();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ dedupKeys.clear();
|
|
|
+ if (pendingSopKeys.size() >= PENDING_KEYS_WARN_THRESHOLD
|
|
|
+ && pendingKeysWarned.compareAndSet(false, true)) {
|
|
|
+ log.warn("{} 待处理关联键已达{}，请注意内存占用", LOG_TAG, pendingSopKeys.size());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 当前 oldAppId 第 1 轮优先，完成后历史 oldAppId 并行（均替换为本次 newAppId）。
|
|
|
+ */
|
|
|
+ private void processAllSopPasses(List<WatchLogKeyVo> allKeys, List<SopPass> passes,
|
|
|
+ String startTime, long epoch,
|
|
|
+ List<Future<long[]>> sopFutures)
|
|
|
+ throws InterruptedException {
|
|
|
+ if (passes.isEmpty()) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ int totalPasses = passes.size();
|
|
|
+
|
|
|
+ runSingleSopPass(allKeys, passes.get(0), startTime, epoch, sopFutures, 1, totalPasses);
|
|
|
+
|
|
|
+ if (totalPasses > 1) {
|
|
|
+ List<SopPass> historicalPasses = new ArrayList<>(passes.subList(1, totalPasses));
|
|
|
+ if (historicalPasses.size() >= 2) {
|
|
|
+ processHistoricalPassesParallel(allKeys, historicalPasses, startTime, epoch,
|
|
|
+ sopFutures, totalPasses);
|
|
|
+ } else {
|
|
|
+ runSingleSopPass(allKeys, historicalPasses.get(0), startTime, epoch, sopFutures, 2, totalPasses);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ progressSopPassLabel = "";
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 历史 oldAppId 有限并行（当前轮已完成后执行） */
|
|
|
+ private void processHistoricalPassesParallel(List<WatchLogKeyVo> allKeys, List<SopPass> historicalPasses,
|
|
|
+ String startTime, long epoch,
|
|
|
+ List<Future<long[]>> sopFutures,
|
|
|
+ int totalPasses)
|
|
|
+ throws InterruptedException {
|
|
|
+ int totalHist = historicalPasses.size();
|
|
|
+ List<Future<?>> coordinators = new ArrayList<>(totalHist);
|
|
|
+
|
|
|
+ for (int i = 0; i < totalHist; i++) {
|
|
|
+ final SopPass pass = historicalPasses.get(i);
|
|
|
+ final int passNo = i + 2;
|
|
|
+ checkRunning(epoch);
|
|
|
+ try {
|
|
|
+ coordinators.add(historicalPassExecutor.submit(() -> {
|
|
|
+ try {
|
|
|
+ runSingleSopPass(allKeys, pass, startTime, epoch, sopFutures, passNo, totalPasses);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ } catch (Exception e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 历史轮次异常 | {} → {}", LOG_TAG, pass.oldAppId, pass.targetNewAppId, e);
|
|
|
+ }
|
|
|
+ }));
|
|
|
+ } catch (RejectedExecutionException e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 历史轮次提交失败，跳过 | {} → {}", LOG_TAG, pass.oldAppId, pass.targetNewAppId, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for (Future<?> coordinator : coordinators) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ awaitCoordinatorFuture(coordinator);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 等待历史轮次协调 Future，带超时与取消，避免协调线程永久阻塞 */
|
|
|
+ private void awaitCoordinatorFuture(Future<?> coordinator) {
|
|
|
+ long deadlineMs = System.currentTimeMillis() + HISTORICAL_PASS_WAIT_SECONDS * 1000L;
|
|
|
+ while (!coordinator.isDone()) {
|
|
|
+ long remainMs = deadlineMs - System.currentTimeMillis();
|
|
|
+ if (remainMs <= 0) {
|
|
|
+ coordinator.cancel(true);
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 历史轮次等待超过{}秒，已取消", LOG_TAG, HISTORICAL_PASS_WAIT_SECONDS);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ coordinator.get(Math.min(remainMs, SOP_FUTURE_WAIT_SECONDS * 1000L), TimeUnit.MILLISECONDS);
|
|
|
+ return;
|
|
|
+ } catch (TimeoutException ignored) {
|
|
|
+ // 继续等待
|
|
|
+ } catch (CancellationException ignored) {
|
|
|
+ return;
|
|
|
+ } catch (ExecutionException e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 历史轮次并行协调失败", LOG_TAG, e.getCause());
|
|
|
+ return;
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ coordinator.cancel(true);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 执行单轮 SOP（一个 oldAppId → 一个 newAppId）；轮内批任务并行，轮末 await 收尾。
|
|
|
+ */
|
|
|
+ private void runSingleSopPass(List<WatchLogKeyVo> allKeys, SopPass pass,
|
|
|
+ String startTime, long epoch,
|
|
|
+ List<Future<long[]>> sopFutures,
|
|
|
+ int passNo, int totalPasses)
|
|
|
+ throws InterruptedException {
|
|
|
+ checkRunning(epoch);
|
|
|
+ progressPhase = PHASE_SOP_PASS;
|
|
|
+ progressSopPassLabel = passNo + "/" + totalPasses + " "
|
|
|
+ + pass.oldAppId + "→" + pass.targetNewAppId
|
|
|
+ + (pass.historical ? "（历史）" : "（当前）");
|
|
|
+ long passStartMs = System.currentTimeMillis();
|
|
|
+ long needBefore = totalNeed.sum();
|
|
|
+ long updatedBefore = totalUpdated.sum();
|
|
|
+
|
|
|
+ log.info("{} SOP第{}/{}轮开始 | [{}] {} → {} | 关联键{}个",
|
|
|
+ LOG_TAG, passNo, totalPasses,
|
|
|
+ pass.historical ? "历史" : "当前优先",
|
|
|
+ pass.oldAppId, pass.targetNewAppId, allKeys.size());
|
|
|
+
|
|
|
+ List<Future<long[]>> passFutures = Collections.synchronizedList(new ArrayList<>());
|
|
|
+ submitSopBatchesForPass(allKeys, pass, startTime, epoch, passFutures);
|
|
|
+ sopFutures.addAll(passFutures);
|
|
|
+
|
|
|
+ progressPhase = PHASE_SOP_DRAIN;
|
|
|
+ awaitAllSopFutures(passFutures, epoch);
|
|
|
+
|
|
|
+ long passNeed = totalNeed.sum() - needBefore;
|
|
|
+ long passUpdated = totalUpdated.sum() - updatedBefore;
|
|
|
+ long passCostSec = (System.currentTimeMillis() - passStartMs) / 1000;
|
|
|
+ String phaseLabel = pass.historical ? "历史" : "当前优先";
|
|
|
+ String status = passNeed == passUpdated
|
|
|
+ ? (passNeed == 0 ? "完成(无待改)" : "完成")
|
|
|
+ : "部分失败";
|
|
|
+
|
|
|
+ log.info("{} SOP第{}/{}轮完成 | [{}] {} → {} | 需处理{} 已处理{} 差额{} | 耗时{} | {}",
|
|
|
+ LOG_TAG, passNo, totalPasses, phaseLabel, pass.oldAppId, pass.targetNewAppId,
|
|
|
+ passNeed, passUpdated, passNeed - passUpdated, formatDuration(passCostSec), status);
|
|
|
+
|
|
|
+ synchronized (taskSopPassSummaries) {
|
|
|
+ taskSopPassSummaries.add(SopPassSummary.executed(pass, phaseLabel, passNeed, passUpdated,
|
|
|
+ passCostSec, status));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 单轮 SOP：send_mini_app_id 等号匹配，批内并行、受背压限制 */
|
|
|
+ private void submitSopBatchesForPass(List<WatchLogKeyVo> allKeys, SopPass pass,
|
|
|
+ String startTime, long epoch,
|
|
|
+ List<Future<long[]>> passFutures)
|
|
|
+ throws InterruptedException {
|
|
|
+ String oldPattern = "\"miniprogramAppid\":\"" + pass.oldAppId + "\"";
|
|
|
+ String newPattern = "\"miniprogramAppid\":\"" + pass.targetNewAppId + "\"";
|
|
|
+
|
|
|
+ Semaphore permits = taskSopPermits;
|
|
|
+ for (int i = 0; i < allKeys.size(); i += SOP_KEY_BATCH_SIZE) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ if (permits != null && !tryAcquireSopPermit(permits, epoch)) {
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ int end = Math.min(i + SOP_KEY_BATCH_SIZE, allKeys.size());
|
|
|
+ List<WatchLogKeyVo> chunk = new ArrayList<>(allKeys.subList(i, end));
|
|
|
+ statSopTasksInFlight.increment();
|
|
|
+ try {
|
|
|
+ Future<long[]> future = sopExecutor.submit(() -> {
|
|
|
+ try {
|
|
|
+ long[] stats = processKeyChunkForAppId(chunk, pass.oldAppId, oldPattern, newPattern,
|
|
|
+ pass.targetNewAppId, startTime, epoch);
|
|
|
+ totalNeed.add(stats[0]);
|
|
|
+ totalUpdated.add(stats[1]);
|
|
|
+ return stats;
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ return new long[]{0, 0};
|
|
|
+ } catch (Exception e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 改库任务出错 | 轮次{}→{} | 本批{}个关联键",
|
|
|
+ LOG_TAG, pass.oldAppId, pass.targetNewAppId, chunk.size(), e);
|
|
|
+ return new long[]{0, 0};
|
|
|
+ } finally {
|
|
|
+ statSopTasksInFlight.decrement();
|
|
|
+ if (permits != null) {
|
|
|
+ permits.release();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ });
|
|
|
+ passFutures.add(future);
|
|
|
+ } catch (RejectedExecutionException e) {
|
|
|
+ statSopTasksInFlight.decrement();
|
|
|
+ if (permits != null) {
|
|
|
+ permits.release();
|
|
|
+ }
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 改库任务提交失败，丢弃本批 | 轮次{}→{} | 关联键{}个",
|
|
|
+ LOG_TAG, pass.oldAppId, pass.targetNewAppId, chunk.size(), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 等待本轮 SOP Future 收尾；单批超时过久则 cancel 丢弃，避免协调线程卡死 */
|
|
|
+ private void awaitAllSopFutures(List<Future<long[]>> sopFutures, long epoch) throws InterruptedException {
|
|
|
+ for (Future<long[]> future : sopFutures) {
|
|
|
+ int timeoutRounds = 0;
|
|
|
+ while (!future.isDone()) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ try {
|
|
|
+ future.get(SOP_FUTURE_WAIT_SECONDS, TimeUnit.SECONDS);
|
|
|
+ } catch (TimeoutException ignored) {
|
|
|
+ timeoutRounds++;
|
|
|
+ if (timeoutRounds >= SOP_FUTURE_MAX_TIMEOUT_ROUNDS) {
|
|
|
+ future.cancel(true);
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 改库任务等待超过{}秒，已取消并丢弃本批",
|
|
|
+ LOG_TAG, SOP_FUTURE_WAIT_SECONDS * timeoutRounds);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ } catch (CancellationException ignored) {
|
|
|
+ break;
|
|
|
+ } catch (ExecutionException e) {
|
|
|
+ statPartitionErrors.increment();
|
|
|
+ log.error("{} 改库任务失败，已跳过本批", LOG_TAG, e.getCause());
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 单 oldAppId 查 SOP（send_mini_app_id = ?，不用 AppId IN）并批量更新。
|
|
|
+ *
|
|
|
+ * @return long[2]：{需更新数, 实际成功数}
|
|
|
+ */
|
|
|
+ private long[] processKeyChunkForAppId(List<WatchLogKeyVo> keys, String oldAppId,
|
|
|
+ String oldPattern, String newPattern,
|
|
|
+ String targetNewAppId, String startTime, long epoch)
|
|
|
+ throws InterruptedException {
|
|
|
+ if (keys.isEmpty() || StringUtils.isBlank(oldAppId)) {
|
|
|
+ return new long[]{0, 0};
|
|
|
+ }
|
|
|
+ checkRunning(epoch);
|
|
|
+
|
|
|
+ List<QwSopLogs> matched;
|
|
|
+ try {
|
|
|
+ matched = sopLogsMapper.selectByKeysForAppId(keys, startTime, oldAppId);
|
|
|
+ } catch (Exception e) {
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 查SOP失败，丢弃本批 | oldAppId={} 关联键{}个", LOG_TAG, oldAppId, keys.size(), e);
|
|
|
+ return new long[]{0, 0};
|
|
|
+ }
|
|
|
+ if (CollectionUtils.isEmpty(matched)) {
|
|
|
+ return new long[]{0, 0};
|
|
|
+ }
|
|
|
+ if (matched.size() > 50_000) {
|
|
|
+ log.info("{} 单批SOP命中{}条偏多 oldAppId={}", LOG_TAG, matched.size(), oldAppId);
|
|
|
+ }
|
|
|
+ statSopMatched.add(matched.size());
|
|
|
+
|
|
|
+ long need = 0;
|
|
|
+ long updated = 0;
|
|
|
+ List<QwSopLogs> updateBatch = new ArrayList<>(UPDATE_BATCH_SIZE);
|
|
|
+
|
|
|
+ for (int offset = 0; offset < matched.size(); offset += 10_000) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ int end = Math.min(offset + 10_000, matched.size());
|
|
|
+ for (int i = offset; i < end; i++) {
|
|
|
+ QwSopLogs sop = matched.get(i);
|
|
|
+ if (sop == null || StringUtils.isBlank(sop.getId())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (!oldAppId.equals(sop.getSendMiniAppId())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String contentJson = sop.getContentJson();
|
|
|
+ if (StringUtils.isBlank(contentJson) || contentJson.length() > MAX_CONTENT_JSON_LENGTH) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (!contentJson.contains(oldPattern)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (contentJson.contains(newPattern)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ need++;
|
|
|
+ needUpdateSopIds.add(sop.getId());
|
|
|
+ updateBatch.add(sop);
|
|
|
+ if (updateBatch.size() >= UPDATE_BATCH_SIZE) {
|
|
|
+ updated += batchUpdateWithRetry(updateBatch, targetNewAppId, epoch);
|
|
|
+ updateBatch.clear();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ matched.clear();
|
|
|
+
|
|
|
+ if (!updateBatch.isEmpty()) {
|
|
|
+ updated += batchUpdateWithRetry(updateBatch, targetNewAppId, epoch);
|
|
|
+ updateBatch.clear();
|
|
|
+ }
|
|
|
+ return new long[]{need, updated};
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 构建更新对象并分批提交；失败指数退避重试。
|
|
|
+ * <p>send_status=3 / memo=更换小程序 为业务约定，使 SOP 重新进入可发送状态。</p>
|
|
|
+ */
|
|
|
+ private int batchUpdateWithRetry(List<QwSopLogs> logs, String newAppId, long epoch)
|
|
|
+ throws InterruptedException {
|
|
|
+ if (CollectionUtils.isEmpty(logs)) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<QwSopLogs> preparedBatch = new ArrayList<>(logs.size());
|
|
|
+ for (QwSopLogs original : logs) {
|
|
|
+ if (original == null || StringUtils.isBlank(original.getId())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String oldAppId = original.getSendMiniAppId();
|
|
|
+ if (StringUtils.isBlank(oldAppId)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ QwSopLogs updateObj = new QwSopLogs();
|
|
|
+ updateObj.setId(original.getId());
|
|
|
+ updateObj.setSendMiniAppId(newAppId);
|
|
|
+ updateObj.setContentJson(replaceAppIdInJson(original.getContentJson(), oldAppId, newAppId));
|
|
|
+ updateObj.setSendStatus(3L);
|
|
|
+ updateObj.setRealSendTime(null);
|
|
|
+ updateObj.setRemark(null);
|
|
|
+ updateObj.setMemo("更换小程序补发");
|
|
|
+ updateObj.setAppSendStatus(0);
|
|
|
+ updateObj.setAppSendRemark(null);
|
|
|
+ preparedBatch.add(updateObj);
|
|
|
+ }
|
|
|
+ if (preparedBatch.isEmpty()) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ int total = 0;
|
|
|
+ for (int i = 0; i < preparedBatch.size(); i += UPDATE_BATCH_SIZE) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ int end = Math.min(i + UPDATE_BATCH_SIZE, preparedBatch.size());
|
|
|
+ List<QwSopLogs> batch = new ArrayList<>(preparedBatch.subList(i, end));
|
|
|
+
|
|
|
+ Exception lastEx = null;
|
|
|
+ int updatedCount = 0;
|
|
|
+ for (int attempt = 1; attempt <= MAX_RETRY; attempt++) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ try {
|
|
|
+ updatedCount = sopLogsMapper.batchUpdateQwSopLogsByIdInMiniAppId(batch);
|
|
|
+ break;
|
|
|
+ } catch (Exception e) {
|
|
|
+ lastEx = e;
|
|
|
+ if (attempt < MAX_RETRY) {
|
|
|
+ long delay = INIT_RETRY_DELAY_MS * (1L << (attempt - 1));
|
|
|
+ Thread.sleep(delay);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (updatedCount == 0 && lastEx != null) {
|
|
|
+ statBatchesDiscarded.increment();
|
|
|
+ log.error("{} 批量写入SOP失败，丢弃本批{}条", LOG_TAG, batch.size(), lastEx);
|
|
|
+ } else if (updatedCount >= batch.size()) {
|
|
|
+ for (QwSopLogs item : batch) {
|
|
|
+ if (item != null && StringUtils.isNotBlank(item.getId())) {
|
|
|
+ successUpdateSopIds.add(item.getId());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else if (updatedCount > 0) {
|
|
|
+ log.info("{} 批量写入SOP仅成功{}/{}条，未成功的id计入漏改", LOG_TAG, updatedCount, batch.size());
|
|
|
+ }
|
|
|
+ total += updatedCount;
|
|
|
+ }
|
|
|
+ return total;
|
|
|
+ }
|
|
|
+
|
|
|
+ // ======================== 工具方法 ========================
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 替换 content_json 中 miniprogramAppid 字段值。
|
|
|
+ * 优先字符串 replace（快）；失败降级 Jackson 解析（兼容格式异常）。
|
|
|
+ */
|
|
|
+ private String replaceAppIdInJson(String contentJson, String oldAppId, String newAppId) {
|
|
|
+ if (StringUtils.isBlank(contentJson) || contentJson.length() > MAX_CONTENT_JSON_LENGTH
|
|
|
+ || StringUtils.isAnyBlank(oldAppId, newAppId)) {
|
|
|
+ return contentJson;
|
|
|
+ }
|
|
|
+ String target = "\"miniprogramAppid\":\"" + oldAppId + "\"";
|
|
|
+ if (!contentJson.contains(target)) {
|
|
|
+ return contentJson;
|
|
|
+ }
|
|
|
+ String replacement = "\"miniprogramAppid\":\"" + newAppId + "\"";
|
|
|
+ try {
|
|
|
+ return contentJson.replace(target, replacement);
|
|
|
+ } catch (Exception e) {
|
|
|
+ try {
|
|
|
+ com.fasterxml.jackson.databind.JsonNode root = objectMapper.readTree(contentJson);
|
|
|
+ com.fasterxml.jackson.databind.JsonNode appIdNode = root.get("miniprogramAppid");
|
|
|
+ if (appIdNode != null && !appIdNode.isNull()
|
|
|
+ && oldAppId.equals(appIdNode.asText())) {
|
|
|
+ ((ObjectNode) root).put("miniprogramAppid", newAppId);
|
|
|
+ return objectMapper.writeValueAsString(root);
|
|
|
+ }
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("{} JSON解析降级失败", LOG_TAG, ex);
|
|
|
+ }
|
|
|
+ return contentJson;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 从看课记录构建 SOP 关联键，与 qw_sop_logs 的 (qw_user_key,sop_id,fs_user_id,external_id) 对应。
|
|
|
+ * 任一必填字段缺失则返回 null（跳过该行，不中断批次）。
|
|
|
+ */
|
|
|
+ private WatchLogKeyVo buildKeyOrNull(FsCourseWatchLog watchLog) {
|
|
|
+ if (watchLog == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ String qwUserId = StringUtils.trimToNull(String.valueOf(watchLog.getQwUserId()));
|
|
|
+ String sopId = StringUtils.trimToNull(watchLog.getSopId());
|
|
|
+ Long userId = watchLog.getUserId();
|
|
|
+ Long externalId = watchLog.getQwExternalContactId();
|
|
|
+ if (StringUtils.isAnyBlank(qwUserId, sopId) || userId == null || externalId == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return new WatchLogKeyVo(qwUserId, sopId, userId, externalId);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 看课游标查询，失败指数退避重试；仍失败返回 null 供上层丢弃本批 */
|
|
|
+ private List<FsCourseWatchLog> fetchWatchLogsWithRetry(String startTime, long cursor, long endId) {
|
|
|
+ Exception lastEx = null;
|
|
|
+ for (int attempt = 1; attempt <= MAX_RETRY; attempt++) {
|
|
|
+ try {
|
|
|
+ return watchLogMapper.selectWatchLogsByCursorUpTo(
|
|
|
+ startTime, cursor, CURSOR_BATCH_SIZE, endId);
|
|
|
+ } catch (Exception e) {
|
|
|
+ lastEx = e;
|
|
|
+ log.warn("{} 看课查询失败 | 第{}次 cursor={}", LOG_TAG, attempt, cursor, e);
|
|
|
+ if (attempt < MAX_RETRY) {
|
|
|
+ try {
|
|
|
+ Thread.sleep(INIT_RETRY_DELAY_MS * (1L << (attempt - 1)));
|
|
|
+ } catch (InterruptedException ie) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.error("{} 看课查询连续失败 cursor={}", LOG_TAG, cursor, lastEx);
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 背压等待 SOP 槽位；累计超时则放弃本批，避免永久阻塞 */
|
|
|
+ private boolean tryAcquireSopPermit(Semaphore permits, long epoch) throws InterruptedException {
|
|
|
+ long waitStart = System.currentTimeMillis();
|
|
|
+ while (!permits.tryAcquire(SOP_PERMIT_WAIT_SECONDS, TimeUnit.SECONDS)) {
|
|
|
+ checkRunning(epoch);
|
|
|
+ long waitedSec = (System.currentTimeMillis() - waitStart) / 1000;
|
|
|
+ log.warn("{} 等待SOP槽位已{}秒 | 在途SOP={}",
|
|
|
+ LOG_TAG, waitedSec, statSopTasksInFlight.sum());
|
|
|
+ if (waitedSec >= SOP_PERMIT_MAX_WAIT_SECONDS) {
|
|
|
+ log.error("{} 等待SOP槽位超过{}秒，丢弃本批", LOG_TAG, SOP_PERMIT_MAX_WAIT_SECONDS);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 大数据量或多段时更快拉满并行窗口；小数据量渐进 ramp，避免 CPU 尖峰 */
|
|
|
+ private static int resolveMaxInFlight(int alreadySubmitted, int partitionCount, long rowCount) {
|
|
|
+ if (partitionCount <= 1) {
|
|
|
+ return 1;
|
|
|
+ }
|
|
|
+ if (rowCount >= SINGLE_PARTITION_FALLBACK_THRESHOLD || partitionCount >= 6) {
|
|
|
+ if (alreadySubmitted < 4) {
|
|
|
+ return Math.min(PARTITION_SUBMIT_WINDOW, 4);
|
|
|
+ }
|
|
|
+ return PARTITION_SUBMIT_WINDOW;
|
|
|
+ }
|
|
|
+ if (alreadySubmitted < 2) {
|
|
|
+ return 2;
|
|
|
+ }
|
|
|
+ if (alreadySubmitted < 6) {
|
|
|
+ return 4;
|
|
|
+ }
|
|
|
+ if (alreadySubmitted < 12) {
|
|
|
+ return 6;
|
|
|
+ }
|
|
|
+ return PARTITION_SUBMIT_WINDOW;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 小数据量全程错峰；大数据/多段仅前几次轻量错峰 */
|
|
|
+ private static boolean shouldStaggerPartitionSubmit(int submittedCount, int partitionCount, long rowCount) {
|
|
|
+ if (rowCount >= SINGLE_PARTITION_FALLBACK_THRESHOLD || partitionCount >= 6) {
|
|
|
+ return submittedCount <= 2 && submittedCount % PARTITION_SUBMIT_STAGGER_EVERY == 0;
|
|
|
+ }
|
|
|
+ return submittedCount <= 8 && submittedCount % PARTITION_SUBMIT_STAGGER_EVERY == 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static int computeTargetPartitionCount(long rowCount) {
|
|
|
+ return (int) Math.min(MAX_PARTITIONS,
|
|
|
+ Math.max(1L, (rowCount + ROWS_PER_PARTITION - 1) / ROWS_PER_PARTITION));
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 多段时一律按行数均分（ROW_NUMBER），保证每段真实行数接近，避免 log_id 空洞导致空段或拖尾段。
|
|
|
+ * 均分失败：大数据量降级为单段；小数据量才回退 log_id 等距。
|
|
|
+ */
|
|
|
+ private List<Partition> buildPartitions(long minId, long maxId, long rowCount, String startTime) {
|
|
|
+ int targetPartitions = computeTargetPartitionCount(rowCount);
|
|
|
+ if (targetPartitions <= 1) {
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("{} 共{}条，按行数均分{}段（协调线程单次查询）…", LOG_TAG, rowCount, targetPartitions);
|
|
|
+ List<Partition> byRow = buildPartitionsByRowStep(minId, maxId, rowCount, startTime, targetPartitions);
|
|
|
+ if (byRow != null && byRow.size() >= 2) {
|
|
|
+ List<Partition> normalized = normalizePartitions(byRow, minId, maxId);
|
|
|
+ logPartitionSpanHint(normalized, rowCount, "按行数均分");
|
|
|
+ return normalized;
|
|
|
+ }
|
|
|
+
|
|
|
+ log.warn("{} 按行数均分失败", LOG_TAG);
|
|
|
+ if (rowCount >= SINGLE_PARTITION_FALLBACK_THRESHOLD) {
|
|
|
+ log.warn("{} 数据量较大，不回退log_id等距（易产生空段/拖尾），改为单段处理", LOG_TAG);
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+ List<Partition> byId = normalizePartitions(
|
|
|
+ buildPartitionsByIdRange(minId, maxId, rowCount), minId, maxId);
|
|
|
+ logPartitionSpanHint(byId, rowCount, "log_id等距");
|
|
|
+ return byId;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** ROW_NUMBER 按约 ROWS_PER_PARTITION 行取分界 log_id，使各段行数接近 */
|
|
|
+ private List<Partition> buildPartitionsByRowStep(long minId, long maxId, long rowCount,
|
|
|
+ String startTime, int targetPartitions) {
|
|
|
+ if (targetPartitions <= 1) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ long stepSize = Math.max(1L, (rowCount + targetPartitions - 1) / targetPartitions);
|
|
|
+ List<Long> boundaryIds = null;
|
|
|
+ Exception lastEx = null;
|
|
|
+ for (int attempt = 1; attempt <= 2; attempt++) {
|
|
|
+ try {
|
|
|
+ long splitStartMs = System.currentTimeMillis();
|
|
|
+ boundaryIds = watchLogMapper.selectLogIdsAtRowStep(startTime, stepSize, rowCount);
|
|
|
+ log.info("{} 按行数均分查询完成 | 第{}次 | 用时{}秒 | 分界点{}个",
|
|
|
+ LOG_TAG, attempt, (System.currentTimeMillis() - splitStartMs) / 1000,
|
|
|
+ boundaryIds == null ? 0 : boundaryIds.size());
|
|
|
+ break;
|
|
|
+ } catch (Exception e) {
|
|
|
+ lastEx = e;
|
|
|
+ log.warn("{} 按行数均分查询失败 | 第{}次", LOG_TAG, attempt, e);
|
|
|
+ if (attempt < 2) {
|
|
|
+ try {
|
|
|
+ Thread.sleep(2000L);
|
|
|
+ } catch (InterruptedException ie) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (boundaryIds == null && lastEx != null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ if (CollectionUtils.isEmpty(boundaryIds)) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ List<Partition> partitions = new ArrayList<>(boundaryIds.size() + 1);
|
|
|
+ long segmentStart = minId;
|
|
|
+ for (Long boundaryId : boundaryIds) {
|
|
|
+ if (boundaryId == null || boundaryId <= segmentStart) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ partitions.add(new Partition(segmentStart, boundaryId));
|
|
|
+ segmentStart = boundaryId + 1;
|
|
|
+ }
|
|
|
+ if (segmentStart <= maxId) {
|
|
|
+ partitions.add(new Partition(segmentStart, maxId));
|
|
|
+ }
|
|
|
+ return partitions.isEmpty() ? null : partitions;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 合并无效/重叠段，去掉 startId>endId 的空段 */
|
|
|
+ private static List<Partition> normalizePartitions(List<Partition> partitions, long minId, long maxId) {
|
|
|
+ if (CollectionUtils.isEmpty(partitions)) {
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+ List<Partition> result = new ArrayList<>(partitions.size());
|
|
|
+ long cursor = minId;
|
|
|
+ for (Partition p : partitions) {
|
|
|
+ if (p == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ long start = Math.max(cursor, p.startId);
|
|
|
+ long end = Math.min(maxId, p.endId);
|
|
|
+ if (start > end) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ result.add(new Partition(start, end));
|
|
|
+ cursor = end + 1;
|
|
|
+ }
|
|
|
+ if (result.isEmpty()) {
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+ Partition last = result.get(result.size() - 1);
|
|
|
+ if (last.endId < maxId && cursor <= maxId) {
|
|
|
+ result.add(new Partition(cursor, maxId));
|
|
|
+ }
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void logPartitionSpanHint(List<Partition> partitions, long rowCount, String strategy) {
|
|
|
+ if (partitions.size() <= 1) {
|
|
|
+ log.info("{} 分段策略={} | 共{}段", LOG_TAG, strategy, partitions.size());
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ long minSpan = Long.MAX_VALUE;
|
|
|
+ long maxSpan = 0;
|
|
|
+ for (Partition p : partitions) {
|
|
|
+ long span = p.endId - p.startId + 1;
|
|
|
+ minSpan = Math.min(minSpan, span);
|
|
|
+ maxSpan = Math.max(maxSpan, span);
|
|
|
+ }
|
|
|
+ long avgRowsHint = rowCount / partitions.size();
|
|
|
+ log.info("{} 分段策略={} | 共{}段 | log_id跨度 min={} max={} | 每段约{}行",
|
|
|
+ LOG_TAG, strategy, partitions.size(), minSpan, maxSpan, avgRowsHint);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 按 log_id 等距切分（O(1) 无 DB 重查询）。
|
|
|
+ * log_id 有空洞时段内行数可能严重不均，仅在小数据量或均分失败时使用。
|
|
|
+ */
|
|
|
+ private List<Partition> buildPartitionsByIdRange(long minId, long maxId, long rowCount) {
|
|
|
+ int targetPartitions = computeTargetPartitionCount(rowCount);
|
|
|
+ if (targetPartitions <= 1) {
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+ long idSpan = maxId - minId + 1;
|
|
|
+ if (idSpan <= targetPartitions) {
|
|
|
+ return Collections.singletonList(new Partition(minId, maxId));
|
|
|
+ }
|
|
|
+ long idStep = Math.max(1L, idSpan / targetPartitions);
|
|
|
+ List<Partition> partitions = new ArrayList<>(targetPartitions);
|
|
|
+ long segmentStart = minId;
|
|
|
+ for (int i = 0; i < targetPartitions - 1; i++) {
|
|
|
+ long segmentEnd = Math.min(maxId, segmentStart + idStep - 1);
|
|
|
+ if (segmentEnd >= maxId) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ partitions.add(new Partition(segmentStart, segmentEnd));
|
|
|
+ segmentStart = segmentEnd + 1;
|
|
|
+ }
|
|
|
+ if (segmentStart <= maxId) {
|
|
|
+ partitions.add(new Partition(segmentStart, maxId));
|
|
|
+ }
|
|
|
+ return partitions.isEmpty()
|
|
|
+ ? Collections.singletonList(new Partition(minId, maxId))
|
|
|
+ : partitions;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 任务开始时根据数据量粗算预计完成时刻（后续进度汇报会按实际速度修正） */
|
|
|
+ private static long estimateInitialFinishMs(long taskStartMs, long rowCount) {
|
|
|
+ if (rowCount <= 0) {
|
|
|
+ return taskStartMs;
|
|
|
+ }
|
|
|
+ long scanRowsPerMin = Math.max(20_000L, PARALLELISM * EST_SCAN_ROWS_PER_THREAD_PER_MIN * 8 / 10);
|
|
|
+ long scanMinutes = (rowCount + scanRowsPerMin - 1) / scanRowsPerMin;
|
|
|
+ long sopExtraMin = Math.max(5L, (rowCount + 500_000L - 1) / 500_000L);
|
|
|
+ return taskStartMs + (scanMinutes + sopExtraMin) * 60_000L;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 统一取消检查点：cancelled / epoch 过期 / 线程 interrupt 均抛出 InterruptedException。
|
|
|
+ * 各层 catch 后应恢复 interrupt 状态或向上传播。
|
|
|
+ */
|
|
|
+ private void checkRunning(long epoch) throws InterruptedException {
|
|
|
+ if (cancelled.get() || epoch != taskEpoch.get() || Thread.currentThread().isInterrupted()) {
|
|
|
+ throw new InterruptedException("任务被取消");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 启动进度汇报：每 5 分钟 1 条（启动时不打，避免与「开始处理」重复） */
|
|
|
+ private ScheduledFuture<?> startProgressHeartbeat(long epoch, long taskStartMs) {
|
|
|
+ synchronized (heartbeatLock) {
|
|
|
+ stopProgressHeartbeat(activeHeartbeatFuture);
|
|
|
+ progressEpoch = epoch;
|
|
|
+ progressTaskStartMs = taskStartMs;
|
|
|
+ progressPhase = PHASE_PREPARE;
|
|
|
+ progressTotalRows = 0L;
|
|
|
+ progressDonePartitions = 0;
|
|
|
+ progressTotalPartitions = 0;
|
|
|
+ heartbeatLastScannedRows = 0L;
|
|
|
+ heartbeatLastTickMs = taskStartMs;
|
|
|
+ progressEstimatedFinishMs = 0L;
|
|
|
+ activeHeartbeatFuture = heartbeatExecutor.scheduleWithFixedDelay(
|
|
|
+ () -> runProgressHeartbeatTick(epoch),
|
|
|
+ PROGRESS_LOG_INTERVAL_MS,
|
|
|
+ PROGRESS_LOG_INTERVAL_MS,
|
|
|
+ TimeUnit.MILLISECONDS);
|
|
|
+ return activeHeartbeatFuture;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void runProgressHeartbeatTick(long epoch) {
|
|
|
+ try {
|
|
|
+ if (destroyed.get() || epoch != progressEpoch || epoch != taskEpoch.get()) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ logProgressHeartbeat();
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("{} 进度汇报线程异常", LOG_TAG, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void stopProgressHeartbeat(ScheduledFuture<?> heartbeatFuture) {
|
|
|
+ progressEpoch = -1L;
|
|
|
+ synchronized (heartbeatLock) {
|
|
|
+ ScheduledFuture<?> toCancel = heartbeatFuture != null ? heartbeatFuture : activeHeartbeatFuture;
|
|
|
+ if (toCancel != null) {
|
|
|
+ toCancel.cancel(false);
|
|
|
+ }
|
|
|
+ if (toCancel == activeHeartbeatFuture) {
|
|
|
+ activeHeartbeatFuture = null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 进度汇报：每 5 分钟输出已处理/目标/百分比，以及改库情况 */
|
|
|
+ private void logProgressHeartbeat() {
|
|
|
+ long nowMs = System.currentTimeMillis();
|
|
|
+ long runSec = progressTaskStartMs > 0 ? (nowMs - progressTaskStartMs) / 1000 : 0;
|
|
|
+ long scannedRows = statWatchLogRows.sum();
|
|
|
+ long totalRows = progressTotalRows;
|
|
|
+ int rowPct = totalRows > 0 ? (int) Math.min(100, scannedRows * 100 / totalRows) : 0;
|
|
|
+ long need = totalNeed.sum();
|
|
|
+ long updated = totalUpdated.sum();
|
|
|
+ long updateGap = need - updated;
|
|
|
+ long sopBusy = statSopTasksInFlight.sum();
|
|
|
+
|
|
|
+ updateEstimatedFinish(nowMs, runSec, scannedRows, totalRows, sopBusy);
|
|
|
+
|
|
|
+ String note = "";
|
|
|
+ if (scannedRows >= totalRows && sopBusy > 0 && StringUtils.isNotBlank(progressSopPassLabel)) {
|
|
|
+ note = " | SOP轮次：" + progressSopPassLabel;
|
|
|
+ } else if (statBatchesDiscarded.sum() > 0) {
|
|
|
+ note = " | 已丢弃批次：" + statBatchesDiscarded.sum();
|
|
|
+ }
|
|
|
+ heartbeatLastScannedRows = scannedRows;
|
|
|
+ heartbeatLastTickMs = nowMs;
|
|
|
+
|
|
|
+ if (totalRows <= 0) {
|
|
|
+ log.info("{} 进度汇报（每5分钟）| 准备中 | 已用{} | 预计完成{}",
|
|
|
+ LOG_TAG, formatDuration(runSec), formatDateTime(progressEstimatedFinishMs));
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("{} 进度汇报（每5分钟）| 看课：已处理{} / 目标{}（{}%）| 改库：需改{} 已改{} 差额{} | 已用{} | 预计完成{}{}",
|
|
|
+ LOG_TAG,
|
|
|
+ scannedRows, totalRows, rowPct,
|
|
|
+ need, updated, updateGap,
|
|
|
+ formatDuration(runSec),
|
|
|
+ formatDateTime(progressEstimatedFinishMs),
|
|
|
+ note);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 按当前均速修正预计完成时刻（运行满 1 分钟后生效） */
|
|
|
+ private void updateEstimatedFinish(long nowMs, long runSec, long scannedRows, long totalRows, long sopBusy) {
|
|
|
+ if (totalRows <= 0) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (runSec < 60) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (scannedRows >= totalRows) {
|
|
|
+ progressEstimatedFinishMs = sopBusy > 0 ? nowMs + PROGRESS_LOG_INTERVAL_MS : nowMs;
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (scannedRows > 0) {
|
|
|
+ long remainSec = (totalRows - scannedRows) * runSec / scannedRows;
|
|
|
+ long sopBufferSec = Math.max(300L, totalRows / 500_000L * 60L);
|
|
|
+ progressEstimatedFinishMs = nowMs + (remainSec + sopBufferSec) * 1000L;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String formatDateTime(long epochMs) {
|
|
|
+ return DATETIME_FMT.format(Instant.ofEpochMilli(epochMs));
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String formatDuration(long seconds) {
|
|
|
+ if (seconds < 60) {
|
|
|
+ return seconds + "秒";
|
|
|
+ }
|
|
|
+ long minutes = seconds / 60;
|
|
|
+ if (minutes < 60) {
|
|
|
+ return minutes + "分钟";
|
|
|
+ }
|
|
|
+ long hours = minutes / 60;
|
|
|
+ long remainMin = minutes % 60;
|
|
|
+ return remainMin > 0 ? hours + "小时" + remainMin + "分钟" : hours + "小时";
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 任务结束报告：预计 vs 实际，一目了然 */
|
|
|
+ private void logEnd(String oldAppId, String newAppId, String startTime,
|
|
|
+ long startMs, String status, String reason) {
|
|
|
+ long endMs = System.currentTimeMillis();
|
|
|
+ long costSec = (endMs - startMs) / 1000;
|
|
|
+ long scanned = statWatchLogRows.sum();
|
|
|
+ long need = totalNeed.sum();
|
|
|
+ long updated = totalUpdated.sum();
|
|
|
+ long estimatedMs = progressEstimatedFinishMs;
|
|
|
+
|
|
|
+ log.info("{} ===== 任务报告 =====", LOG_TAG);
|
|
|
+ log.info("{} 结果：{}{}", LOG_TAG, status,
|
|
|
+ StringUtils.isNotBlank(reason) ? "（" + reason + "）" : "");
|
|
|
+ log.info("{} 换绑：{} → {} | 数据日期：{}", LOG_TAG, oldAppId, newAppId, startTime);
|
|
|
+ log.info("{} 开始：{}", LOG_TAG, formatDateTime(startMs));
|
|
|
+ log.info("{} 结束：{}", LOG_TAG, formatDateTime(endMs));
|
|
|
+ log.info("{} 预计完成：{} | 实际耗时：{}", LOG_TAG,
|
|
|
+ estimatedMs > 0 ? formatDateTime(estimatedMs) : formatDateTime(endMs),
|
|
|
+ formatDuration(costSec));
|
|
|
+ if (estimatedMs > 0 && "完成".equals(status)) {
|
|
|
+ long diffSec = Math.abs(endMs - estimatedMs) / 1000;
|
|
|
+ if (diffSec <= 60) {
|
|
|
+ log.info("{} 与预计对比：基本准时（误差{}分钟内）", LOG_TAG, Math.max(1, diffSec / 60));
|
|
|
+ } else {
|
|
|
+ String diffDesc = endMs <= estimatedMs
|
|
|
+ ? "比预计早" + formatDuration(diffSec)
|
|
|
+ : "比预计晚" + formatDuration(diffSec);
|
|
|
+ log.info("{} 与预计对比：{}", LOG_TAG, diffDesc);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.info("{} 看课扫描：{}条", LOG_TAG, scanned);
|
|
|
+ log.info("{} 改库：需改{}条 | 已改{}条 | 差额{}条", LOG_TAG, need, updated, need - updated);
|
|
|
+ long discarded = statBatchesDiscarded.sum();
|
|
|
+ if (discarded > 0) {
|
|
|
+ log.info("{} 丢弃批次：{}（超时或异常，详见上方错误日志）", LOG_TAG, discarded);
|
|
|
+ }
|
|
|
+ long errors = statPartitionErrors.sum();
|
|
|
+ if (errors > 0) {
|
|
|
+ log.info("{} 异常计数：{}", LOG_TAG, errors);
|
|
|
+ }
|
|
|
+ logSopPassDetailReport();
|
|
|
+ logUpdateGapConclusion(need, updated);
|
|
|
+ log.info("{} ====================", LOG_TAG);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void logSopPassDetailReport() {
|
|
|
+ List<SopPassSummary> summaries;
|
|
|
+ synchronized (taskSopPassSummaries) {
|
|
|
+ if (taskSopPassSummaries.isEmpty()) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ summaries = new ArrayList<>(taskSopPassSummaries);
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("{} --- 各旧小程序换绑明细 ---", LOG_TAG);
|
|
|
+ int partialFailed = 0;
|
|
|
+ for (SopPassSummary s : summaries) {
|
|
|
+ long gap = s.need - s.updated;
|
|
|
+ if (gap != 0) {
|
|
|
+ partialFailed++;
|
|
|
+ }
|
|
|
+ log.info("{} [{}] {} → {} | 需处理{} 已处理{} 差额{} | 耗时{} | {}",
|
|
|
+ LOG_TAG, s.phaseLabel, s.oldAppId, s.targetNewAppId,
|
|
|
+ s.need, s.updated, gap, formatDuration(s.durationSec), s.status);
|
|
|
+ }
|
|
|
+
|
|
|
+ String coverage = partialFailed == 0
|
|
|
+ ? "Redis登记" + taskRegistryOldAppCount + "个旧AppId均已执行，需处理=已处理"
|
|
|
+ : "有" + partialFailed + "个旧AppId存在未改完记录，请核查";
|
|
|
+ log.info("{} 覆盖结论：共执行{}轮 | 登记{}个旧AppId | {}",
|
|
|
+ LOG_TAG, summaries.size(), taskRegistryOldAppCount, coverage);
|
|
|
+ log.info("{} --- 明细结束 ---", LOG_TAG);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void logUpdateGapConclusion(long need, long updated) {
|
|
|
+ long gap = need - updated;
|
|
|
+ if (gap == 0) {
|
|
|
+ log.info("{} 结论：数据处理全部成功", LOG_TAG);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (gap > 0) {
|
|
|
+ Set<String> missed = computeMissedSopIds();
|
|
|
+ log.info("{} 结论：有{}条数据未补发", LOG_TAG, gap);
|
|
|
+ logMissedSopIds(missed);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ log.info("{} 结论：已改超出需改，请核查", LOG_TAG);
|
|
|
+ }
|
|
|
+
|
|
|
+ private Set<String> computeMissedSopIds() {
|
|
|
+ Set<String> missed = new HashSet<>(needUpdateSopIds);
|
|
|
+ missed.removeAll(successUpdateSopIds);
|
|
|
+ return missed;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 分批打印漏改的 qw_sop_logs.id，避免单条日志过长 */
|
|
|
+ private void logMissedSopIds(Set<String> missedIds) {
|
|
|
+ if (missedIds.isEmpty()) {
|
|
|
+ log.info("{} 漏改qw_sop_logs.id：未能定位具体记录，请按差额在库内核查", LOG_TAG);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ List<String> ids = new ArrayList<>(missedIds);
|
|
|
+ Collections.sort(ids);
|
|
|
+ int logCount = Math.min(ids.size(), MAX_MISSED_IDS_LOG);
|
|
|
+ for (int i = 0; i < logCount; i += 100) {
|
|
|
+ int end = Math.min(i + 100, logCount);
|
|
|
+ log.info("{} 漏改qw_sop_logs.id：{}", LOG_TAG, String.join(",", ids.subList(i, end)));
|
|
|
+ }
|
|
|
+ if (ids.size() > MAX_MISSED_IDS_LOG) {
|
|
|
+ log.info("{} 漏改id共{}条，以上仅展示前{}条", LOG_TAG, ids.size(), MAX_MISSED_IDS_LOG);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 单轮 SOP 查改：历史 oldAppId 统一替换为本次 newAppId */
|
|
|
+ private static final class SopPass {
|
|
|
+ final String oldAppId;
|
|
|
+ final String targetNewAppId;
|
|
|
+ /** true=历史 oldAppId 轮，false=当前 oldAppId（始终第 1 轮） */
|
|
|
+ final boolean historical;
|
|
|
+
|
|
|
+ SopPass(String oldAppId, String targetNewAppId, boolean historical) {
|
|
|
+ this.oldAppId = oldAppId;
|
|
|
+ this.targetNewAppId = targetNewAppId;
|
|
|
+ this.historical = historical;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 单个 oldAppId 处理结果 */
|
|
|
+ private static final class SopPassSummary {
|
|
|
+ final String oldAppId;
|
|
|
+ final String targetNewAppId;
|
|
|
+ final String phaseLabel;
|
|
|
+ final long need;
|
|
|
+ final long updated;
|
|
|
+ final long durationSec;
|
|
|
+ final String status;
|
|
|
+
|
|
|
+ private SopPassSummary(String oldAppId, String targetNewAppId, String phaseLabel,
|
|
|
+ long need, long updated, long durationSec, String status) {
|
|
|
+ this.oldAppId = oldAppId;
|
|
|
+ this.targetNewAppId = targetNewAppId;
|
|
|
+ this.phaseLabel = phaseLabel;
|
|
|
+ this.need = need;
|
|
|
+ this.updated = updated;
|
|
|
+ this.durationSec = durationSec;
|
|
|
+ this.status = status;
|
|
|
+ }
|
|
|
+
|
|
|
+ static SopPassSummary executed(SopPass pass, String phaseLabel,
|
|
|
+ long need, long updated, long durationSec, String status) {
|
|
|
+ return new SopPassSummary(pass.oldAppId, pass.targetNewAppId, phaseLabel,
|
|
|
+ need, updated, durationSec, status);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** log_id 闭区间 [startId, endId] */
|
|
|
+ private static class Partition {
|
|
|
+ final long startId;
|
|
|
+ final long endId;
|
|
|
+
|
|
|
+ Partition(long start, long end) {
|
|
|
+ this.startId = start;
|
|
|
+ this.endId = end;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static ThreadFactory namedThreadFactory(final String prefix) {
|
|
|
+ final AtomicLong seq = new AtomicLong(1);
|
|
|
+ return r -> {
|
|
|
+ Thread t = new Thread(r, prefix + "-" + seq.getAndIncrement());
|
|
|
+ t.setDaemon(true);
|
|
|
+ t.setUncaughtExceptionHandler((thread, ex) ->
|
|
|
+ log.error("{} 线程未捕获异常 thread={}", LOG_TAG, thread.getName(), ex));
|
|
|
+ return t;
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private static void shutdownExecutor(ExecutorService executor, String name, long awaitSeconds) {
|
|
|
+ executor.shutdown();
|
|
|
+ try {
|
|
|
+ if (!executor.awaitTermination(awaitSeconds, TimeUnit.SECONDS)) {
|
|
|
+ executor.shutdownNow();
|
|
|
+ if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
|
|
|
+ log.info("{} {} 未能在限定时间内完全终止", LOG_TAG, name);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ executor.shutdownNow();
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ======================== 生命周期：Spring 容器关闭时回收线程池 ========================
|
|
|
+
|
|
|
+ @PreDestroy
|
|
|
+ public void destroy() {
|
|
|
+ shutdown();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 优雅关闭：递增 epoch → 取消分区 → 先停协调池(60s) 再停分区池(30s)。
|
|
|
+ * 重复调用安全（CAS destroyed）。
|
|
|
+ */
|
|
|
+ public void shutdown() {
|
|
|
+ if (destroyed.compareAndSet(false, true)) {
|
|
|
+ taskEpoch.incrementAndGet();
|
|
|
+ cancelled.set(true);
|
|
|
+ cancelActiveFutures();
|
|
|
+ stopProgressHeartbeat(null);
|
|
|
+ shutdownExecutor(historicalPassExecutor, "miniapp-replace-hist-pass", 30);
|
|
|
+ shutdownExecutor(sopExecutor, "miniapp-replace-sop", 60);
|
|
|
+ shutdownExecutor(heartbeatExecutor, "miniapp-replace-heartbeat", 5);
|
|
|
+ shutdownExecutor(partitionExecutor, "miniapp-replace-partition", 30);
|
|
|
+ shutdownExecutor(coordinatorExecutor, "miniapp-replace-coordinator", 60);
|
|
|
+ log.info("{} 补发小程序线程池已销毁", LOG_TAG);
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
