feat(站内通知): 麦象异常扫描上线——任务停滞/失败/队列积压/批量空结果推管理员
用户要求「maixiang异常也要发通知」，确认范围（卡住/失败/队列堆积/服务不可用）与 受众（只发管理员）后实现： - MaixiangConsoleClient：18960 后台只读接口客户端（batch/tasks、tasks、queue/status、 batch/detail），token 鉴权、失败只记日志。URL 必须用 URI 对象提交——字符串形态会被 RestClient 二次编码（%20→%2520），end_time 参数实测报 Incorrect DATETIME value， 检查会静默失效；已加回归单测 buildUrlEncodesTimeParameterExactlyOnce - MaixiangAnomalyScanner 五类检查：①批量任务停滞（status 0/1 超时无更新，默认 60 分钟） ②单任务滞留（创建超 30 分钟未完成）③近 30 分钟失败达最小条数 ④批量任务空结果 （worker 报错时批次仍被标"已完成"，是唯一可抓的批量失败信号）⑤队列积压（pending≥300 / processing≥100）；去重按天/小时/任务，全部为管理员全局事件（subjectUserId=null） - 接入 NotificationScanScheduler（分布式锁内同跑）；scene=maixiang_anomaly； AIIMAGE_NOTIFICATION_MAIXIANG_* 环境变量可调，令牌留空=跳过 - 前端 NotificationScene 类型补 maixiang_anomaly（铃铛不按 scene 渲染，无运行时改动） - 测试：客户端解析（真实抓包样例）+ 扫描器 14 例 + 手工联调探针 MaixiangLiveProbeTest （-Dmaixiang.live=true 开启，只读不写通知） 已部署双节点（.env 加 console token、JAR cd3e06c4）并线上验证：首扫推 2×15 条管理员通知 （3 个变体任务停滞 / 39 个跟价任务滞留），二次扫描落库=0 去重生效。
This commit is contained in:
@@ -41,6 +41,30 @@ public class NotificationProperties {
|
||||
/** jikip 代理接口探测开关（复用余额查询接口判活，使用 user-secret 的 jikip 配置）。 */
|
||||
private boolean jikipProbeEnabled = true;
|
||||
|
||||
/** 麦象（18960 任务调度）异常扫描开关：任务停滞/失败/队列积压 → 管理员通知。 */
|
||||
private boolean maixiangScanEnabled = true;
|
||||
|
||||
/**
|
||||
* 麦象后台接口令牌（18960 console token）。留空=跳过麦象异常扫描——
|
||||
* 该令牌与「跟价任务 API 地址」(priceTrackApiUrl) 一起构成后台只读接口的访问凭据。
|
||||
*/
|
||||
private String maixiangConsoleToken = "";
|
||||
|
||||
/** 麦象批量任务停滞阈值（分钟）：status=0/1 且超过该时长无更新视为卡住。 */
|
||||
private int maixiangStuckMinutes = 60;
|
||||
|
||||
/** 麦象单任务滞留阈值（分钟）：创建超时仍未完成（status=0/1）视为滞留/无人消费。 */
|
||||
private int maixiangSingleStuckMinutes = 30;
|
||||
|
||||
/** 麦象任务失败告警阈值（条）：近 30 分钟窗口内失败数达到该值才告警。 */
|
||||
private int maixiangFailMinCount = 1;
|
||||
|
||||
/** 麦象队列积压阈值（条）：task:queue 待处理数达到该值告警。 */
|
||||
private int maixiangQueuePendingThreshold = 300;
|
||||
|
||||
/** 麦象队列积压阈值（条）：task:processing 处理中数达到该值告警。 */
|
||||
private int maixiangQueueProcessingThreshold = 100;
|
||||
|
||||
/** 已读通知保留天数（超期自动清理），默认 90 天。 */
|
||||
private int readRetentionDays = 90;
|
||||
}
|
||||
|
||||
+240
@@ -0,0 +1,240 @@
|
||||
package com.nanri.aiimage.modules.notification.client;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.config.HttpClientPool;
|
||||
import com.nanri.aiimage.config.NotificationProperties;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.client.RestClient;
|
||||
import org.springframework.web.util.UriComponentsBuilder;
|
||||
|
||||
import java.net.URI;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 麦象（18960 任务调度）后台只读接口客户端。
|
||||
*
|
||||
* <p>站内通知的「麦象异常扫描」所需数据全部来自这几个 console 接口（token 鉴权）：
|
||||
* 批量任务列表（all_task）/ 单任务列表（task_record）/ 队列长度 / 批量任务结果行数。
|
||||
*
|
||||
* <p>只读、无副作用；调用失败只记中文日志并返回空结果（不抛异常），
|
||||
* 避免扫描任务被单次网络抖动打断——服务级不可用由现有 18960 探测告警覆盖。
|
||||
* 注意日志里不输出完整 URL（含后台令牌）。
|
||||
*/
|
||||
@Component
|
||||
@Slf4j
|
||||
@RequiredArgsConstructor
|
||||
public class MaixiangConsoleClient {
|
||||
|
||||
private static final int READ_TIMEOUT_MILLIS = 8_000;
|
||||
private static final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||||
/** 麦象后台接口未注册结果表时返回的哨兵值。 */
|
||||
public static final long RESULT_TOTAL_UNSUPPORTED = -1L;
|
||||
|
||||
private final NotificationProperties properties;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
/** 是否已配置（地址 + 后台令牌）；未配置时调用方应跳过扫描。 */
|
||||
public boolean configured() {
|
||||
return hasText(properties.getPriceTrackApiUrl()) && hasText(properties.getMaixiangConsoleToken());
|
||||
}
|
||||
|
||||
/** 批量任务列表（all_task）。createdBefore 非空时只取创建时间不晚于该时刻的任务（SQL 字符串比较）。 */
|
||||
public BatchTaskPage batchTasks(int status, LocalDateTime createdBefore, int pageSize) {
|
||||
String url = buildUrl("/api/console/batch/tasks",
|
||||
"status=" + status,
|
||||
"page=1",
|
||||
"page_size=" + pageSize,
|
||||
createdBefore == null ? null : "end_time=" + TIME_FORMAT.format(createdBefore));
|
||||
return parseBatchTaskPage(fetch(url, "批量任务列表 status=" + status));
|
||||
}
|
||||
|
||||
/** 单任务列表（task_record）。 */
|
||||
public SingleTaskPage singleTasks(int status, LocalDateTime createdBefore, int pageSize) {
|
||||
String url = buildUrl("/api/console/tasks",
|
||||
"status=" + status,
|
||||
"page=1",
|
||||
"page_size=" + pageSize,
|
||||
createdBefore == null ? null : "end_time=" + TIME_FORMAT.format(createdBefore));
|
||||
return parseSingleTaskPage(fetch(url, "单任务列表 status=" + status));
|
||||
}
|
||||
|
||||
/** 单任务队列状态（task:queue:N / task:processing:N，仅返回非零级别）。 */
|
||||
public List<QueueStateItem> queueStatus() {
|
||||
return parseQueueStates(fetch(buildUrl("/api/console/queue/status"), "队列状态"));
|
||||
}
|
||||
|
||||
/** 批量任务结果行数：未注册结果表（table=null）返回 {@link #RESULT_TOTAL_UNSUPPORTED}。 */
|
||||
public long batchResultTotal(String taskId) {
|
||||
String url = buildUrl("/api/console/batch/detail", "task_id=" + taskId, "page=1", "page_size=1");
|
||||
return parseBatchResultTotal(fetch(url, "批量任务结果 task_id=" + taskId));
|
||||
}
|
||||
|
||||
// ======================== HTTP ========================
|
||||
|
||||
/** 拉取并展开统一信封 {code,mes,data}；失败返回 null（已记日志）。 */
|
||||
private JsonNode fetch(String url, String scene) {
|
||||
String body;
|
||||
try {
|
||||
// 用 URI 对象提交：URL 已由 buildUrl 编码过一次，不能再经 RestClient 模板再次编码
|
||||
// （字符串形态会被二次编码，%20 变 %2520，服务端 DATETIME 参数直接报错）
|
||||
body = RestClient.builder()
|
||||
.requestFactory(HttpClientPool.requestFactory(READ_TIMEOUT_MILLIS))
|
||||
.build()
|
||||
.get()
|
||||
.uri(URI.create(url))
|
||||
.retrieve()
|
||||
.body(String.class);
|
||||
} catch (Exception ex) {
|
||||
log.warn("[maixiang-console] 请求失败 scene={} err={}", scene, ex.getMessage());
|
||||
return null;
|
||||
}
|
||||
return parseEnvelope(body, scene);
|
||||
}
|
||||
|
||||
/** 组装后台接口 URL（含 token 与查询参数，统一编码一次；包内可见供单测校验编码行为）。 */
|
||||
String buildUrl(String path, String... queryPairs) {
|
||||
UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(baseUrl())
|
||||
.path(path)
|
||||
.queryParam("token", properties.getMaixiangConsoleToken().trim());
|
||||
for (String pair : queryPairs) {
|
||||
if (pair == null) {
|
||||
continue;
|
||||
}
|
||||
int idx = pair.indexOf('=');
|
||||
builder.queryParam(pair.substring(0, idx), pair.substring(idx + 1));
|
||||
}
|
||||
return builder.build().encode().toUriString();
|
||||
}
|
||||
|
||||
private String baseUrl() {
|
||||
String base = properties.getPriceTrackApiUrl() == null ? "" : properties.getPriceTrackApiUrl().trim();
|
||||
return base.endsWith("/") ? base.substring(0, base.length() - 1) : base;
|
||||
}
|
||||
|
||||
// ======================== 解析（包内可见，供单测用真实样例直接校验） ========================
|
||||
|
||||
/** 展开信封：code=1 返回 data 节点，否则记日志返回 null。 */
|
||||
JsonNode parseEnvelope(String body, String scene) {
|
||||
if (!hasText(body)) {
|
||||
log.warn("[maixiang-console] 响应为空 scene={}", scene);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
JsonNode root = objectMapper.readTree(body);
|
||||
if (root.path("code").asInt(-1) != 1) {
|
||||
log.warn("[maixiang-console] 接口返回失败 scene={} mes={}", scene, root.path("mes").asText(""));
|
||||
return null;
|
||||
}
|
||||
return root.path("data");
|
||||
} catch (Exception ex) {
|
||||
log.warn("[maixiang-console] 响应解析失败 scene={} err={}", scene, ex.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
BatchTaskPage parseBatchTaskPage(JsonNode data) {
|
||||
if (data == null || data.isMissingNode()) {
|
||||
return BatchTaskPage.EMPTY;
|
||||
}
|
||||
List<BatchTaskItem> items = new ArrayList<>();
|
||||
for (JsonNode node : data.path("items")) {
|
||||
items.add(new BatchTaskItem(
|
||||
node.path("task_id").asText(""),
|
||||
node.path("task_type").asText(""),
|
||||
node.path("uid").asText(""),
|
||||
node.path("status").asInt(-1),
|
||||
parseTime(node.path("create_time").asText(null)),
|
||||
parseTime(node.path("update_time").asText(null))));
|
||||
}
|
||||
return new BatchTaskPage(data.path("total").asLong(items.size()), items);
|
||||
}
|
||||
|
||||
SingleTaskPage parseSingleTaskPage(JsonNode data) {
|
||||
if (data == null || data.isMissingNode()) {
|
||||
return SingleTaskPage.EMPTY;
|
||||
}
|
||||
List<SingleTaskItem> items = new ArrayList<>();
|
||||
for (JsonNode node : data.path("items")) {
|
||||
items.add(new SingleTaskItem(
|
||||
node.path("task_id").asText(""),
|
||||
node.path("task_type").asText(""),
|
||||
node.path("task_status").asInt(-1),
|
||||
parseTime(node.path("create_time").asText(null)),
|
||||
parseTime(node.path("update_time").asText(null)),
|
||||
node.path("task_result").asText("")));
|
||||
}
|
||||
return new SingleTaskPage(data.path("total").asLong(items.size()), items);
|
||||
}
|
||||
|
||||
List<QueueStateItem> parseQueueStates(JsonNode data) {
|
||||
List<QueueStateItem> items = new ArrayList<>();
|
||||
if (data == null || data.isMissingNode()) {
|
||||
return items;
|
||||
}
|
||||
for (JsonNode node : data.path("queues")) {
|
||||
items.add(new QueueStateItem(
|
||||
node.path("level").asInt(0),
|
||||
node.path("queue_pending").asLong(0),
|
||||
node.path("queue_processing").asLong(0)));
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
long parseBatchResultTotal(JsonNode data) {
|
||||
if (data == null || data.isMissingNode()) {
|
||||
return RESULT_TOTAL_UNSUPPORTED;
|
||||
}
|
||||
JsonNode table = data.path("table");
|
||||
if (table.isMissingNode() || table.isNull() || !hasText(table.asText(""))) {
|
||||
// 未注册结果表（或表名非法）：该类型不支持按结果行数判断，跳过
|
||||
return RESULT_TOTAL_UNSUPPORTED;
|
||||
}
|
||||
return data.path("total").asLong(0);
|
||||
}
|
||||
|
||||
private LocalDateTime parseTime(String text) {
|
||||
if (!hasText(text)) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return LocalDateTime.parse(text.trim(), TIME_FORMAT);
|
||||
} catch (Exception ex) {
|
||||
log.warn("[maixiang-console] 时间解析失败 text={}", text);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private boolean hasText(String value) {
|
||||
return value != null && !value.trim().isEmpty();
|
||||
}
|
||||
|
||||
// ======================== 数据模型 ========================
|
||||
|
||||
/** 批量任务（all_task）单条。 */
|
||||
public record BatchTaskItem(String taskId, String taskType, String uid, int status,
|
||||
LocalDateTime createTime, LocalDateTime updateTime) {
|
||||
}
|
||||
|
||||
public record BatchTaskPage(long total, List<BatchTaskItem> items) {
|
||||
public static final BatchTaskPage EMPTY = new BatchTaskPage(0, List.of());
|
||||
}
|
||||
|
||||
/** 单任务（task_record）单条。 */
|
||||
public record SingleTaskItem(String taskId, String taskType, int status,
|
||||
LocalDateTime createTime, LocalDateTime updateTime, String resultRaw) {
|
||||
}
|
||||
|
||||
public record SingleTaskPage(long total, List<SingleTaskItem> items) {
|
||||
public static final SingleTaskPage EMPTY = new SingleTaskPage(0, List.of());
|
||||
}
|
||||
|
||||
/** 队列状态一行。 */
|
||||
public record QueueStateItem(int level, long pending, long processing) {
|
||||
}
|
||||
}
|
||||
+312
@@ -0,0 +1,312 @@
|
||||
package com.nanri.aiimage.modules.notification.service;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.config.NotificationProperties;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskItem;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskPage;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.QueueStateItem;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskItem;
|
||||
import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskPage;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 麦象（18960 任务调度）异常扫描：把「采集端出了事但没人知道」的情况推给管理员。
|
||||
*
|
||||
* <p>五类检查（全部只读，通过 {@link MaixiangConsoleClient} 的后台接口取数）：
|
||||
* <ol>
|
||||
* <li>批量任务停滞：status=0/1 且超过阈值无更新（如消费端 worker 掉线中断采集）；</li>
|
||||
* <li>单任务滞留：创建超过阈值仍未完成（跟价等单任务无人消费）；</li>
|
||||
* <li>任务失败：近窗口内 status=3 的单任务达到最小条数；</li>
|
||||
* <li>批量任务空结果：已完成但结果 0 条（worker 中途报错时批次仍会被标"已完成"）；</li>
|
||||
* <li>队列积压：task:queue / task:processing 超过阈值。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>全部为管理员全局事件（与「服务探测」一致，不按数据权限过滤）；去重键：
|
||||
* 停滞类按天、失败/积压按小时、空结果按任务——同一异常窗口内只提醒一次，
|
||||
* 情况变化（新增停滞任务/失败数增长）时刷新内容并重新提醒。
|
||||
*
|
||||
* <p>由 {@link NotificationScanScheduler} 在分布式锁内调用；单项检查异常只记日志不中断其余检查。
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class MaixiangAnomalyScanner {
|
||||
|
||||
private static final String SCENE = NotificationService.SCENE_MAIXIANG_ANOMALY;
|
||||
|
||||
/** 失败聚合窗口（分钟）：与扫描间隔解耦，窗口内失败数达到阈值才告警。 */
|
||||
private static final int FAIL_WINDOW_MINUTES = 30;
|
||||
/** 空结果检查窗口（分钟）：只看最近完成的批量任务（覆盖数个扫描周期）。 */
|
||||
private static final int BATCH_EMPTY_WINDOW_MINUTES = 15;
|
||||
/** 单类接口单轮拉取上限（停滞类任务量级很小，100 足够覆盖）。 */
|
||||
private static final int PAGE_SIZE = 100;
|
||||
private static final int FAIL_PAGE_SIZE = 50;
|
||||
/** 通知内容里的任务号预览条数。 */
|
||||
private static final int MAX_PREVIEW = 5;
|
||||
private static final int REASON_MAX_LENGTH = 120;
|
||||
|
||||
private static final DateTimeFormatter DAY_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
|
||||
private static final DateTimeFormatter HOUR_FORMAT = DateTimeFormatter.ofPattern("yyyyMMddHH");
|
||||
private static final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||||
|
||||
/** 麦象任务类型 → 中文名（未登记的类型回退原始值）。 */
|
||||
private static final Map<String, String> TYPE_LABELS = Map.ofEntries(
|
||||
Map.entry("price-track-run", "跟价"),
|
||||
Map.entry("variant-collection", "变体采集"),
|
||||
Map.entry("collect-data-run", "采集数据"),
|
||||
Map.entry("collect-data-run-pagels", "采集数据(搜索)"),
|
||||
Map.entry("collect-data-run-detail", "采集数据(详情)"),
|
||||
Map.entry("search-collection", "搜索采集"),
|
||||
Map.entry("shop-data-crawl-run", "店铺数据采集"),
|
||||
Map.entry("similar-asin-run", "货源查询"),
|
||||
Map.entry("appearance-patent-run", "外观专利"));
|
||||
|
||||
private final NotificationProperties properties;
|
||||
private final MaixiangConsoleClient consoleClient;
|
||||
private final NotificationDispatchService notificationDispatchService;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
/** 扫描入口：配置缺失时跳过；各检查互相隔离，单项失败不影响其余。 */
|
||||
public void scan() {
|
||||
if (!properties.isMaixiangScanEnabled()) {
|
||||
log.info("[maixiang-scan] 麦象异常扫描已关闭（maixiang-scan-enabled=false），跳过");
|
||||
return;
|
||||
}
|
||||
if (!consoleClient.configured()) {
|
||||
log.info("[maixiang-scan] 麦象扫描跳过：未配置 price-track-api-url / maixiang-console-token");
|
||||
return;
|
||||
}
|
||||
NotificationDispatchService.AdminAudience audience = notificationDispatchService.prepareAdminAudience();
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
runCheck("批量任务停滞", () -> checkStuckBatchTasks(audience, now));
|
||||
runCheck("单任务滞留", () -> checkStuckSingleTasks(audience, now));
|
||||
runCheck("任务失败", () -> checkFailedSingleTasks(audience, now));
|
||||
runCheck("批量任务空结果", () -> checkEmptyBatchResults(audience, now));
|
||||
runCheck("队列积压", () -> checkQueueBacklog(audience, now));
|
||||
}
|
||||
|
||||
private void runCheck(String name, Runnable check) {
|
||||
try {
|
||||
check.run();
|
||||
} catch (Exception ex) {
|
||||
log.warn("[maixiang-scan] {} 检查异常（不影响其余检查）err={}", name, ex.getMessage(), ex);
|
||||
}
|
||||
}
|
||||
|
||||
/** 批量任务停滞：创建早于阈值、仍处 status=0/1 且 update_time 不再推进。 */
|
||||
void checkStuckBatchTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) {
|
||||
int thresholdMinutes = Math.max(1, properties.getMaixiangStuckMinutes());
|
||||
LocalDateTime cutoff = now.minusMinutes(thresholdMinutes);
|
||||
List<BatchTaskItem> candidates = new ArrayList<>();
|
||||
for (int status : new int[]{0, 1}) {
|
||||
candidates.addAll(consoleClient.batchTasks(status, cutoff, PAGE_SIZE).items());
|
||||
}
|
||||
// 创建时间早于 cutoff 但仍在正常推进的大任务（update_time 晚于 cutoff）不算停滞
|
||||
List<BatchTaskItem> stuck = new ArrayList<>();
|
||||
for (BatchTaskItem task : candidates) {
|
||||
if (task.updateTime() != null && !task.updateTime().isAfter(cutoff)) {
|
||||
stuck.add(task);
|
||||
}
|
||||
}
|
||||
if (stuck.isEmpty()) {
|
||||
log.info("[maixiang-scan] 无停滞的批量任务（阈值 {} 分钟）", thresholdMinutes);
|
||||
return;
|
||||
}
|
||||
LocalDateTime earliest = stuck.stream()
|
||||
.map(BatchTaskItem::updateTime)
|
||||
.min(LocalDateTime::compareTo)
|
||||
.orElse(cutoff);
|
||||
String content = "麦象有 " + stuck.size() + " 个批量任务超过 " + thresholdMinutes
|
||||
+ " 分钟无进展（" + typeSummary(stuck.stream().map(BatchTaskItem::taskType).toList())
|
||||
+ "；任务 " + taskPreview(stuck)
|
||||
+ "；最早停滞于 " + TIME_FORMAT.format(earliest)
|
||||
+ "）。消费端可能已掉线，请检查采集机 worker 是否在运行。";
|
||||
push(audience, "麦象批量任务停滞", content, "maixiang_stuck_batch:" + now.format(DAY_FORMAT));
|
||||
}
|
||||
|
||||
/** 单任务滞留：创建早于阈值、仍处 status=0/1（跟价等单任务应秒级完成）。 */
|
||||
void checkStuckSingleTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) {
|
||||
int thresholdMinutes = Math.max(1, properties.getMaixiangSingleStuckMinutes());
|
||||
LocalDateTime cutoff = now.minusMinutes(thresholdMinutes);
|
||||
List<SingleTaskItem> stuck = new ArrayList<>();
|
||||
for (int status : new int[]{0, 1}) {
|
||||
SingleTaskPage page = consoleClient.singleTasks(status, cutoff, PAGE_SIZE);
|
||||
for (SingleTaskItem task : page.items()) {
|
||||
if (task.updateTime() != null && !task.updateTime().isAfter(cutoff)) {
|
||||
stuck.add(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (stuck.isEmpty()) {
|
||||
log.info("[maixiang-scan] 无滞留的单任务（阈值 {} 分钟）", thresholdMinutes);
|
||||
return;
|
||||
}
|
||||
String content = "麦象有 " + stuck.size() + " 个单任务超过 " + thresholdMinutes
|
||||
+ " 分钟未完成（" + typeSummary(stuck.stream().map(SingleTaskItem::taskType).toList())
|
||||
+ "；任务 " + singleTaskPreview(stuck)
|
||||
+ "）。可能存在任务无人消费或消费端卡住，请检查采集机 worker 与队列消费情况。";
|
||||
push(audience, "麦象任务滞留未消费", content, "maixiang_stuck_single:" + now.format(DAY_FORMAT));
|
||||
}
|
||||
|
||||
/** 任务失败：近窗口内 status=3 的单任务达到最小条数（按小时桶聚合，原因取最新一条）。 */
|
||||
void checkFailedSingleTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) {
|
||||
SingleTaskPage page = consoleClient.singleTasks(3, null, FAIL_PAGE_SIZE);
|
||||
LocalDateTime windowStart = now.minusMinutes(FAIL_WINDOW_MINUTES);
|
||||
List<SingleTaskItem> failed = new ArrayList<>();
|
||||
for (SingleTaskItem task : page.items()) {
|
||||
if (task.updateTime() != null && !task.updateTime().isBefore(windowStart)) {
|
||||
failed.add(task);
|
||||
}
|
||||
}
|
||||
int minCount = Math.max(1, properties.getMaixiangFailMinCount());
|
||||
if (failed.size() < minCount) {
|
||||
log.info("[maixiang-scan] 近 {} 分钟麦象失败 {} 条（未达阈值 {}）", FAIL_WINDOW_MINUTES, failed.size(), minCount);
|
||||
return;
|
||||
}
|
||||
String reason = latestReason(failed);
|
||||
String content = "近 " + FAIL_WINDOW_MINUTES + " 分钟麦象有 " + failed.size() + " 条任务失败（"
|
||||
+ typeSummary(failed.stream().map(SingleTaskItem::taskType).toList()) + "）"
|
||||
+ (reason.isEmpty() ? "" : "；最近失败原因：" + reason)
|
||||
+ "。请排查采集端配置（如代理余额、任务类型注册）。";
|
||||
push(audience, "麦象任务失败", content, "maixiang_fail:" + now.format(HOUR_FORMAT));
|
||||
}
|
||||
|
||||
/** 批量任务空结果：最近完成但结果 0 条（worker 报错时批次仍会被标为"已完成"）。 */
|
||||
void checkEmptyBatchResults(NotificationDispatchService.AdminAudience audience, LocalDateTime now) {
|
||||
BatchTaskPage page = consoleClient.batchTasks(2, null, PAGE_SIZE);
|
||||
LocalDateTime windowStart = now.minusMinutes(BATCH_EMPTY_WINDOW_MINUTES);
|
||||
for (BatchTaskItem task : page.items()) {
|
||||
if (task.updateTime() == null || task.updateTime().isBefore(windowStart)) {
|
||||
continue;
|
||||
}
|
||||
long total = consoleClient.batchResultTotal(task.taskId());
|
||||
if (total != 0) {
|
||||
continue;
|
||||
}
|
||||
String content = "批量任务 " + task.taskId() + "（" + labelOf(task.taskType()) + "，uid=" + task.uid()
|
||||
+ "）已完成但结果 0 条，可能采集中途异常（worker 报错时任务仍会被标为已完成），请核查。";
|
||||
push(audience, "麦象批量任务无结果", content, "maixiang_batch_empty:" + task.taskId());
|
||||
}
|
||||
}
|
||||
|
||||
/** 队列积压：task:queue 待处理 / task:processing 处理中超过阈值。 */
|
||||
void checkQueueBacklog(NotificationDispatchService.AdminAudience audience, LocalDateTime now) {
|
||||
int pendingThreshold = Math.max(1, properties.getMaixiangQueuePendingThreshold());
|
||||
int processingThreshold = Math.max(1, properties.getMaixiangQueueProcessingThreshold());
|
||||
StringBuilder detail = new StringBuilder();
|
||||
for (QueueStateItem queue : consoleClient.queueStatus()) {
|
||||
boolean pendingHit = queue.pending() >= pendingThreshold;
|
||||
boolean processingHit = queue.processing() >= processingThreshold;
|
||||
if (!pendingHit && !processingHit) {
|
||||
continue;
|
||||
}
|
||||
if (detail.length() > 0) {
|
||||
detail.append("；");
|
||||
}
|
||||
detail.append("L").append(queue.level())
|
||||
.append(" 待处理 ").append(queue.pending())
|
||||
.append(" / 处理中 ").append(queue.processing());
|
||||
}
|
||||
if (detail.length() == 0) {
|
||||
log.info("[maixiang-scan] 无队列积压（阈值 待处理{} / 处理中{}）", pendingThreshold, processingThreshold);
|
||||
return;
|
||||
}
|
||||
String content = "麦象单任务队列积压：" + detail + "（阈值 待处理 " + pendingThreshold
|
||||
+ " / 处理中 " + processingThreshold + "）。消费端可能掉线或处理变慢，请尽快排查。";
|
||||
push(audience, "麦象队列积压", content, "maixiang_queue_backlog:" + now.format(HOUR_FORMAT));
|
||||
}
|
||||
|
||||
// ======================== 文案辅助 ========================
|
||||
|
||||
private void push(NotificationDispatchService.AdminAudience audience, String title, String content, String dedupeKeyBase) {
|
||||
int pushed = notificationDispatchService.pushToAdmins(audience, SCENE, NotificationService.LEVEL_WARNING,
|
||||
"麦象异常：" + title, content, dedupeKeyBase, null);
|
||||
log.info("[maixiang-scan] 已推送管理员通知 title={} 落库={} 条 dedupe={}", title, pushed, dedupeKeyBase);
|
||||
}
|
||||
|
||||
/** 类型分布摘要，例如「变体采集 3 个、跟价 1 个」。 */
|
||||
private String typeSummary(List<String> taskTypes) {
|
||||
Map<String, Integer> counts = new LinkedHashMap<>();
|
||||
for (String type : taskTypes) {
|
||||
counts.merge(labelOf(type), 1, Integer::sum);
|
||||
}
|
||||
StringBuilder summary = new StringBuilder();
|
||||
for (Map.Entry<String, Integer> entry : counts.entrySet()) {
|
||||
if (summary.length() > 0) {
|
||||
summary.append("、");
|
||||
}
|
||||
summary.append(entry.getKey()).append(" ").append(entry.getValue()).append(" 个");
|
||||
}
|
||||
return summary.toString();
|
||||
}
|
||||
|
||||
private String taskPreview(List<BatchTaskItem> tasks) {
|
||||
return preview(tasks.stream().map(BatchTaskItem::taskId).toList(), tasks.size());
|
||||
}
|
||||
|
||||
private String singleTaskPreview(List<SingleTaskItem> tasks) {
|
||||
return preview(tasks.stream().map(SingleTaskItem::taskId).toList(), tasks.size());
|
||||
}
|
||||
|
||||
private String preview(List<String> ids, int totalCount) {
|
||||
List<String> shown = new ArrayList<>(MAX_PREVIEW);
|
||||
for (String id : ids) {
|
||||
if (id == null || id.isBlank()) {
|
||||
continue;
|
||||
}
|
||||
shown.add(id);
|
||||
if (shown.size() >= MAX_PREVIEW) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (shown.isEmpty()) {
|
||||
return "无";
|
||||
}
|
||||
String text = String.join("、", shown);
|
||||
return totalCount > shown.size() ? text + " 等" : text;
|
||||
}
|
||||
|
||||
/** 取最新一条失败原因：task_result 为 JSON（如 {"error": "..."}），解析失败退化为原文截断。 */
|
||||
private String latestReason(List<SingleTaskItem> failed) {
|
||||
for (SingleTaskItem task : failed) {
|
||||
String raw = task.resultRaw();
|
||||
if (raw == null || raw.isBlank()) {
|
||||
continue;
|
||||
}
|
||||
String reason = raw;
|
||||
try {
|
||||
JsonNode node = objectMapper.readTree(raw);
|
||||
if (node.path("error").isTextual()) {
|
||||
reason = node.path("error").asText();
|
||||
}
|
||||
} catch (Exception ignore) {
|
||||
// 非 JSON：直接用原文
|
||||
}
|
||||
reason = reason.replaceAll("\\s+", " ").trim();
|
||||
if (reason.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
return reason.length() <= REASON_MAX_LENGTH ? reason : reason.substring(0, REASON_MAX_LENGTH) + "…";
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
private String labelOf(String taskType) {
|
||||
String type = taskType == null ? "" : taskType.trim();
|
||||
if (type.isEmpty()) {
|
||||
return "未知类型";
|
||||
}
|
||||
return TYPE_LABELS.getOrDefault(type, type);
|
||||
}
|
||||
}
|
||||
+6
@@ -35,6 +35,9 @@ import java.util.Map;
|
||||
* <p>服务探测：品牌检测服务（15126） / 跟价任务 API（18960） / jikip 代理接口，
|
||||
* 失败立即重试一次（过滤瞬抖），两次都失败才告警；同服务每小时最多一条。
|
||||
*
|
||||
* <p>麦象异常扫描：见 {@link MaixiangAnomalyScanner}（批量/单任务停滞、任务失败、
|
||||
* 批量任务空结果、队列积压 → 管理员通知），与任务失败扫描共用本调度与分布式锁。
|
||||
*
|
||||
* <p>双实例经 Redis 分布式锁互斥；所有分支留中文日志便于线上排查。
|
||||
*/
|
||||
@Slf4j
|
||||
@@ -67,6 +70,7 @@ public class NotificationScanScheduler {
|
||||
private final DistributedJobLockService distributedJobLockService;
|
||||
private final NotificationProperties properties;
|
||||
private final ProxyBalancePort proxyBalancePort;
|
||||
private final MaixiangAnomalyScanner maixiangAnomalyScanner;
|
||||
|
||||
/** 上次清理日期（每天最多清理一次；双节点由分布式锁保证只有一个实例执行）。 */
|
||||
private volatile LocalDate lastCleanupDate;
|
||||
@@ -90,6 +94,8 @@ public class NotificationScanScheduler {
|
||||
if (properties.isServiceProbeEnabled()) {
|
||||
probeServices();
|
||||
}
|
||||
// 麦象异常扫描（任务停滞/失败/队列积压）；开关与凭据在扫描器内部判定
|
||||
maixiangAnomalyScanner.scan();
|
||||
cleanupExpiredIfNeeded();
|
||||
} catch (Exception ex) {
|
||||
log.warn("[notification-scan] 扫描异常终止 err={}", ex.getMessage(), ex);
|
||||
|
||||
+2
@@ -36,6 +36,8 @@ public class NotificationService {
|
||||
public static final String SCENE_SECRET_INVALID = "secret_invalid";
|
||||
public static final String SCENE_TASK_FAILED = "task_failed";
|
||||
public static final String SCENE_SERVICE_DOWN = "service_down";
|
||||
/** 麦象（18960 任务调度）异常：任务停滞/失败/队列积压。 */
|
||||
public static final String SCENE_MAIXIANG_ANOMALY = "maixiang_anomaly";
|
||||
public static final String SCENE_SYSTEM = "system";
|
||||
|
||||
private static final long MAX_PAGE_SIZE = 100L;
|
||||
|
||||
@@ -354,6 +354,15 @@ aiimage:
|
||||
brand-service-url: ${AIIMAGE_NOTIFICATION_BRAND_SERVICE_URL:}
|
||||
price-track-api-url: ${AIIMAGE_NOTIFICATION_PRICE_TRACK_API_URL:}
|
||||
jikip-probe-enabled: ${AIIMAGE_NOTIFICATION_JIKIP_PROBE_ENABLED:true}
|
||||
# 麦象（18960 任务调度）异常扫描：批量/单任务停滞、任务失败、队列积压 → 管理员通知。
|
||||
# console-token 留空=跳过麦象扫描；接口地址复用上面的 price-track-api-url。
|
||||
maixiang-scan-enabled: ${AIIMAGE_NOTIFICATION_MAIXIANG_SCAN_ENABLED:true}
|
||||
maixiang-console-token: ${AIIMAGE_NOTIFICATION_MAIXIANG_CONSOLE_TOKEN:}
|
||||
maixiang-stuck-minutes: ${AIIMAGE_NOTIFICATION_MAIXIANG_STUCK_MINUTES:60}
|
||||
maixiang-single-stuck-minutes: ${AIIMAGE_NOTIFICATION_MAIXIANG_SINGLE_STUCK_MINUTES:30}
|
||||
maixiang-fail-min-count: ${AIIMAGE_NOTIFICATION_MAIXIANG_FAIL_MIN_COUNT:1}
|
||||
maixiang-queue-pending-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PENDING_THRESHOLD:300}
|
||||
maixiang-queue-processing-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PROCESSING_THRESHOLD:100}
|
||||
read-retention-days: ${AIIMAGE_NOTIFICATION_READ_RETENTION_DAYS:90}
|
||||
security:
|
||||
shop-credential-key: ${AIIMAGE_SHOP_CREDENTIAL_KEY:change-me-shop-credential-key}
|
||||
|
||||
Reference in New Issue
Block a user