# DailyHotJob 改造任务清单 ## 改造目标 修复 DailyHotJob 异常吞没、性能低下、运维盲区等核心缺陷，改造为可观测、可中断、高性能的定时任务。 ## 涉及文件清单 | 文件 | 路径 | 改动量 | |------|------|--------| | DailyHotJob.java | `yudao-module-wellness/.../wellness/job/jackett/DailyHotJob.java` | 小改 | | JackettServiceImpl.java | `yudao-module-wellness/.../wellness/service/jackett/impl/JackettServiceImpl.java` | 大改 | | ResourcesUtils.java | `yudao-module-wellness/.../wellness/utils/ResourcesUtils.java` | 小改 | | application-{env}.yml | `yudao-module-wellness/.../resources/` | 新增配置项 | --- ## 分步任务 ### P0-1：异常不再吞没，让 JobHandlerInvoker 能记录真实状态 **描述**：当前 `DailyHotJob.execute()` 没有 throws Exception，内部 `dailyHot()` 所有异常都被 try-catch 吞掉。JobHandlerInvoker 永远收不到异常，Job 日志永远显示成功，运维完全盲区。 **改动点**： 1. **`DailyHotJob.java`**：`execute()` 方法改为 `throws Exception`，不再包 try-catch ```java @Override public String execute(String param) throws Exception { jackettService.dailyHot(); return "每日热点"; } ``` 注意：如果 `dailyHot()` 声明了 throws Exception，这里会自动向上传播，不需要额外处理。 2. **`JackettServiceImpl.java` — `dailyHot()` 方法签名**：改为 `throws Exception` ```java @Override @TenantIgnore public void dailyHot() throws Exception { ... } ``` 3. **`dailyHot()` 内部循环**：**保留**单个关键词搜索失败的捕获（因为一个搜失败不应该影响其他关键词），但去掉最外层那层兜底 try-catch，或者让最外层 catch 后重新 throw： ```java // ❌ 错误做法：catch 后只打日志 // ✅ 正确做法：catch 可恢复异常后继续循环，不可恢复异常向上抛 for (String javId : newIds) { try { ... } catch (Exception e) { log.error("搜索 '{}' 时出错", javId, e.getMessage(), e); // 单个失败不中断整体任务，continue 继续下一个 } } ``` 这样单个关键词搜索失败不会影响整体，但 `dailyHot()` 本身如果出现致命错误（如 FlareSolverr 不可用），会在 `getPageContent()` 抛出异常并向上传播。 **注意事项**： - `dailyHot()` 的循环内 `catch` 后应该 `continue`，而不是 `break` 或 `throw` - 如果 `ResourcesUtils.saveFile()` 抛出 `RuntimeException`，它也会向上传播——需要考虑这个是否属于「可恢复」异常 --- ### P0-2：租户注解问题修正 **描述**：`@TenantIgnore` 只在 `dailyHot()` 上，但内部调用的 `getDailyHotIds()`、`getPageContent()`、`searchSingleKeyword()`、`searchRecent()` 都是通过 `this.` 内部调用，Spring AOP 切面不会触发，租户 SQL 拦截器可能拼入错误的 `WHERE tenant_id = ?`。 **改动点**： 1. 在这些方法上补 `@TenantIgnore`： - `getDailyHotIds()` - `getPageContent(String targetUrl)` - `searchSingleKeyword(String query, Integer limit)` - `searchRecent(List keywords, Integer hours, Integer limit)` ```java @Override @TenantIgnore public Set getDailyHotIds() { ... } @Override @TenantIgnore public String getPageContent(String targetUrl) { ... } @Override @TenantIgnore public List searchSingleKeyword(String query, Integer limit) { ... } @Override @TenantIgnore public List searchRecent(List keywords, Integer hours, Integer limit) { ... } ``` **注意事项**： - `saveRedisIds()` 也加了 `@TenantIgnore` 了吗？检查一下，如果没有也要加上 - 如果 ruoyi 框架的租户拦截是通过 `@TenantIgnore` 在 Service 方法上加注解来跳过，这些方法都需要加 --- ### P1-1：批量查重 + 批量插入替代 N+1 **描述**：当前对每个 hotId 执行一次 `selectOne`（N 次查询）+ 循环内逐条 `insert`，严重低效。 **改动点**： 1. **`JavInfoMapper.java`**：新增批量查询方法 ```java // 根据 javId 列表批量查询已存在的记录 List selectByJavIds(@Param("javIds") Collection javIds); ``` 对应的 XML： ```xml ``` 2. **`JavInfoMapper.java`**：新增批量插入方法 ```java // 批量插入（忽略重复键） int insertBatch(@Param("list") Collection list); ``` 对应的 XML（使用 MySQL `INSERT IGNORE` 或 `ON DUPLICATE KEY UPDATE`）： ```xml INSERT IGNORE INTO wellness_jav_info(jav_id, create_time, update_time) VALUES (#{item.javId}, #{item.createTime}, #{item.updateTime}) ``` 3. **`JackettServiceImpl.dailyHot()`**：改造查重逻辑 ```java // ❌ 旧：逐条 selectOne Set newIds = hotIds.stream() .filter(javId -> javInfoMapper.selectOne(JavInfoDO::getJavId, javId) == null) .collect(Collectors.toSet()); // ✅ 新：批量查重 Set existingIds = javInfoMapper.selectByJavIds(hotIds) .stream().map(JavInfoDO::getJavId).collect(Collectors.toSet()); Set newIds = hotIds.stream() .filter(id -> !existingIds.contains(id)) .collect(Collectors.toSet()); ``` 4. **`JackettServiceImpl.dailyHot()`**：改造插入逻辑，循环结束后统一批量插入 ```java // 循环内不再逐条 insert，改为收集待插入 List toInsert = new ArrayList<>(); for (String javId : newIds) { // ...搜索+过滤+保存文件... toInsert.add(new JavInfoDO().setJavId(javId)); } // 循环结束后批量插入 if (!toInsert.isEmpty()) { javInfoMapper.insertBatch(toInsert); } ``` **注意事项**： - `insertBatch` 需要 MySQL 支持，如果是 PostgreSQL 或 Oracle 需要对应适配 - 使用 `INSERT IGNORE` 可以避免重复键冲突；如果不需要 `ignore` 行为，可以用 `ON DUPLICATE KEY UPDATE` 更新 update_time --- ### P1-2：并发搜索 + 超时控制 **描述**：逐条串行搜索 + `Thread.sleep`，如果 50 个新 ID，仅 sleep 就 100s，加上 HTTP 耗时 2-5s/次，整个 Job 跑几分钟。 **改动点**： 1. **`JackettServiceImpl.java`**：引入并发控制 ```java import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; ``` 在类中声明线程池（或者使用 Spring 的 `TaskExecutor`）： ```java private static final ExecutorService SEARCH_EXECUTOR = Executors.newFixedThreadPool(5); ``` 改造 `dailyHot()` 中的循环： ```java // ❌ 旧：串行 for 循环 + sleep for (String javId : newIds) { List results = searchSingleKeyword(javId, 30); // ...过滤+保存... Thread.sleep(jackettProperties.getSearchDelay()); } // ✅ 新：CompletableFuture 并发 List> futures = new ArrayList<>(); for (String javId : newIds) { CompletableFuture future = CompletableFuture.runAsync(() -> { try { List results = searchSingleKeyword(javId, 30); // ...过滤+保存... } catch (Exception e) { log.error("搜索 '{}' 时出错", javId, e.getMessage(), e); } }, SEARCH_EXECUTOR); futures.add(future); } // 等待所有搜索完成（带超时） CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .get(jackettProperties.getSearchTimeoutMinutes(), TimeUnit.MINUTES); ``` `searchRecent()` 同理也要改造。 2. **`JackettProperties.java`**：新增配置项 ```java /** 搜索超时时间（分钟） */ private Integer searchTimeoutMinutes = 30; ``` 3. **`application.yml`**：验证或添加配置 ```yaml jackett: search-timeout-minutes: 30 ``` 4. **`RestTemplate` 配置**：在 `JackettConfig` 或 `RestTemplate` 的 `@Bean` 创建处，设置连接超时和读取超时 ```java SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory(); factory.setConnectTimeout(10_000); // 10秒连接超时 factory.setReadTimeout(30_000); // 30秒读取超时 RestTemplate restTemplate = new RestTemplate(factory); ``` **注意事项**： - 线程池要确保应用关闭时正确 shutdown（可以用 `@PreDestroy` 或在 Spring 的 `@Bean(destroyMethod="shutdown")`） - 并发度不宜过高（3-5 即可），避免 Jackett 服务端限流 - `searchSingleKeyword()` 和 `searchRecent()` 内部已有 try-catch，在 CompletableFuture 内还会继续 catch，注意异常不会自动传播到 `future.get()`，需要 `CompletableFuture` 的 `handle()` 或自行封装 --- ### P2-1：文件按日期分片替代无限追加 **描述**：`ResourcesUtils.saveFile(content, wujiBaseDirectory, "hot")` 永远追加到同一个 `hot.txt`，文件无限增长。 **改动点**： 1. **`JackettServiceImpl.dailyHot()`**：传入文件名时带上日期 ```java // ❌ 旧 ResourcesUtils.saveFile(resultVO.getMagnetUri(), jackettProperties.getWujiBaseDirectory(), "hot"); // ✅ 新：按日期分文件 String dateStr = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd")); ResourcesUtils.saveFile(resultVO.getMagnetUri(), jackettProperties.getWujiBaseDirectory(), "hot-" + dateStr); ``` 或按年月日三级目录： ```java String yyyy = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy")); String MMdd = LocalDateTime.now().format(DateTimeFormatter.ofPattern("MM/dd")); ResourcesUtils.saveFile(resultVO.getMagnetUri(), jackettProperties.getWujiBaseDirectory() + "/" + yyyy + "/" + MMdd, "hot"); ``` 2. 或者如果决定改入库（推荐方案）： 在 `JavInfoDO` 中新增 `magnet_uri` 字段，在 `insertBatch` 时一并写入，完全替代文件存储。 此时 `dailyHot()` 循环内去掉 `saveFile` 调用，`toInsert` 中设置 `magnetUri`。 **注意事项**： - 如果选择入库替代文件，需要考虑 `JavInfoDO` 表结构变更和数据库 migration，改动量更大 - 如果继续用文件，建议增加清理策略（如保留最近 30 天的文件，自动删除过期文件） --- ### P2-2：日期过滤改为时间窗口 **描述**：当前用 `resultVO.getPublishDate().getYear() != currentYear` 过滤，语义是「当年发布」而非「最近发布」。跨年时段逻辑偏差明显。 **改动点**： **`JackettServiceImpl.dailyHot()`**：替换过滤条件 ```java // ❌ 旧 if (resultVO.getPublishDate() == null || resultVO.getPublishDate().getYear() != currentYear) { continue; } // ✅ 新：最近 24 小时内发布 LocalDateTime since = LocalDateTime.now().minusHours(24); if (resultVO.getPublishDate() == null || resultVO.getPublishDate().isBefore(since)) { continue; } ``` 时间窗口大小建议外部化配置： ```yaml jackett: hot-window-hours: 24 ``` --- ### P2-3：避免 Thread.sleep 阻塞 Quartz 线程 **描述**：`Thread.sleep(searchDelay)` 在 Quartz 的 `@DisallowConcurrentExecution` 工作线程中阻塞，无法处理其他 Job。 **改动点**： 如果改用了 P1-2 的并发方案（CompletableFuture），线程池已经独立于 Quartz 线程，sleep 问题自然解决。如果暂时不改并发，至少： 1. 将 `searchDelay` 从 sleep 改为使用 `ScheduledExecutorService` 调度 2. 或者在并发方案中就解决了，不用额外改动 **推荐**：P1-2 的并发方案已经涵盖了这一条，不需要独立处理。 --- ### P3-1：Redis Key 硬编码修复 **描述**：`saveRedisIds()` 中的 Redis key 硬编码为 `"\""jav:rank:2023"\""`，年份过时且带有多余引号。 **改动点**： **`JackettProperties.java`**：新增配置项 ```java /** Redis 热门排名 Key */ private String redisRankKey = "jav:rank"; ``` **`JackettServiceImpl.saveRedisIds()`**：改为使用配置 ```java // ❌ 旧 String key = "\""jav:rank:2023"\""; // ✅ 新 String key = jackettProperties.getRedisRankKey(); ``` **`application.yml`**：添加配置 ```yaml jackett: redis-rank-key: "jav:rank:2026" # 或更通用 key ``` --- ### P3-2：幂等性/中断支持 **描述**：Job 执行中断时，已 insert 的记录和已保存的文件产生不一致，下次执行产生重复数据。 **改动点**： 1. **数据库层**：`JavInfoDO.jav_id` 加 UNIQUE 约束，`insertBatch` 使用 `INSERT IGNORE` → 幂等写入，重复自动跳过 ```sql ALTER TABLE wellness_jav_info ADD UNIQUE INDEX uk_jav_id (jav_id); ``` 2. **文件层**：按日期分片后（P2-1），同一天执行多次也是追加到同一个日期文件，仍然会重复。可以在写入前检查是否已存在： ```java // 方式 A：先读文件内容做 Set 去重（小文件可行，大文件低效） // 方式 B：不管去重，反正按日期分片，运维定期清理即可 ``` 3. **「执行中标记」**：可选地在 Redis 中设置一个锁/标记，防止同一个 Job 实例并发（Quartz 的 `@DisallowConcurrentExecution` 已经解决了同一实例的并发，但多实例部署还需要确认） --- ## 验收条件 | # | 验收项 | 验证方式 | |---|--------|----------| | 1 | 异常能被 Job 框架记录 | 手动使 FlareSolverr 不可用，观察 Job 日志面板显示失败状态 | | 2 | 租户注解正确 | 多租户环境下执行 Job，确认没有错误的 `tenant_id` SQL | | 3 | 数据库批量查询+插入 | 100+ hotIds 时，SQL 日志只出现一次 select 和一次 insert | | 4 | 并发搜索 | 50 个新 ID 的搜索时间 ≤ 30s（之前 4+ 分钟） | | 5 | 文件按日期分片 | 执行每天 Job 后，目录下生成 `hot-2026-04-26.txt` 格式文件，而非累加到一个文件 | | 6 | 日期过滤正确 | 插入数据库的应该是最近 24h 发布的数据，而非整年数据 | | 7 | Redis Key 可配置 | 修改 `application.yml` 中的 `jackett.redis-rank-key`，Job 读取新值 | | 8 | 幂等性 | 同一批 hotIds 执行两次，数据库和文件不产生重复记录 |