394 lines
14 KiB
Markdown
394 lines
14 KiB
Markdown
# 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<String> keywords, Integer hours, Integer limit)`
|
||
|
||
```java
|
||
@Override
|
||
@TenantIgnore
|
||
public Set<String> getDailyHotIds() { ... }
|
||
|
||
@Override
|
||
@TenantIgnore
|
||
public String getPageContent(String targetUrl) { ... }
|
||
|
||
@Override
|
||
@TenantIgnore
|
||
public List<JackettSearchResultVO> searchSingleKeyword(String query, Integer limit) { ... }
|
||
|
||
@Override
|
||
@TenantIgnore
|
||
public List<JackettSearchResultVO> searchRecent(List<String> keywords, Integer hours, Integer limit) { ... }
|
||
```
|
||
|
||
**注意事项**：
|
||
- `saveRedisIds()` 也加了 `@TenantIgnore` 了吗？检查一下，如果没有也要加上
|
||
- 如果 ruoyi 框架的租户拦截是通过 `@TenantIgnore` 在 Service 方法上加注解来跳过，这些方法都需要加
|
||
|
||
---
|
||
|
||
### P1-1：批量查重 + 批量插入替代 N+1
|
||
|
||
**描述**：当前对每个 hotId 执行一次 `selectOne`（N 次查询）+ 循环内逐条 `insert`，严重低效。
|
||
|
||
**改动点**：
|
||
|
||
1. **`JavInfoMapper.java`**：新增批量查询方法
|
||
```java
|
||
// 根据 javId 列表批量查询已存在的记录
|
||
List<JavInfoDO> selectByJavIds(@Param("javIds") Collection<String> javIds);
|
||
```
|
||
|
||
对应的 XML：
|
||
```xml
|
||
<select id="selectByJavIds" resultMap="JavInfoResultMap">
|
||
SELECT * FROM wellness_jav_info WHERE jav_id IN
|
||
<foreach collection="javIds" item="javId" open="(" separator="," close=")">
|
||
#{javId}
|
||
</foreach>
|
||
</select>
|
||
```
|
||
|
||
2. **`JavInfoMapper.java`**：新增批量插入方法
|
||
```java
|
||
// 批量插入（忽略重复键）
|
||
int insertBatch(@Param("list") Collection<JavInfoDO> list);
|
||
```
|
||
|
||
对应的 XML（使用 MySQL `INSERT IGNORE` 或 `ON DUPLICATE KEY UPDATE`）：
|
||
```xml
|
||
<insert id="insertBatch">
|
||
INSERT IGNORE INTO wellness_jav_info(jav_id, create_time, update_time) VALUES
|
||
<foreach collection="list" item="item" separator=",">
|
||
(#{item.javId}, #{item.createTime}, #{item.updateTime})
|
||
</foreach>
|
||
</insert>
|
||
```
|
||
|
||
3. **`JackettServiceImpl.dailyHot()`**：改造查重逻辑
|
||
```java
|
||
// ❌ 旧：逐条 selectOne
|
||
Set<String> newIds = hotIds.stream()
|
||
.filter(javId -> javInfoMapper.selectOne(JavInfoDO::getJavId, javId) == null)
|
||
.collect(Collectors.toSet());
|
||
|
||
// ✅ 新：批量查重
|
||
Set<String> existingIds = javInfoMapper.selectByJavIds(hotIds)
|
||
.stream().map(JavInfoDO::getJavId).collect(Collectors.toSet());
|
||
Set<String> newIds = hotIds.stream()
|
||
.filter(id -> !existingIds.contains(id))
|
||
.collect(Collectors.toSet());
|
||
```
|
||
|
||
4. **`JackettServiceImpl.dailyHot()`**：改造插入逻辑，循环结束后统一批量插入
|
||
```java
|
||
// 循环内不再逐条 insert，改为收集待插入
|
||
List<JavInfoDO> 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<JackettSearchResultVO> results = searchSingleKeyword(javId, 30);
|
||
// ...过滤+保存...
|
||
Thread.sleep(jackettProperties.getSearchDelay());
|
||
}
|
||
|
||
// ✅ 新：CompletableFuture 并发
|
||
List<CompletableFuture<Void>> futures = new ArrayList<>();
|
||
for (String javId : newIds) {
|
||
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
|
||
try {
|
||
List<JackettSearchResultVO> 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 执行两次，数据库和文件不产生重复记录 |
|