yangxuan/openclaw-config

Fork 0

Files

T

yangxuan 57f354cf9f 结构调整

2026-07-14 16:36:24 +08:00

14 KiB

Raw Blame History

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 日志永远显示成功，运维完全盲区。

改动点：

DailyHotJob.java：execute() 方法改为 throws Exception，不再包 try-catch
```
@Override
public String execute(String param) throws Exception {
    jackettService.dailyHot();
    return "每日热点";
}
```
注意：如果 dailyHot() 声明了 throws Exception，这里会自动向上传播，不需要额外处理。

JackettServiceImpl.java — dailyHot() 方法签名：改为 throws Exception

@Override
@TenantIgnore
public void dailyHot() throws Exception { ... }

dailyHot() 内部循环：保留单个关键词搜索失败的捕获（因为一个搜失败不应该影响其他关键词），但去掉最外层那层兜底 try-catch，或者让最外层 catch 后重新 throw：

// ❌ 错误做法：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 = ?。

改动点：

在这些方法上补 @TenantIgnore：

getDailyHotIds()
getPageContent(String targetUrl)
searchSingleKeyword(String query, Integer limit)
searchRecent(List<String> keywords, Integer hours, Integer limit)

@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，严重低效。

改动点：

JavInfoMapper.java：新增批量查询方法

// 根据 javId 列表批量查询已存在的记录
List<JavInfoDO> selectByJavIds(@Param("javIds") Collection<String> javIds);

对应的 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>

JavInfoMapper.java：新增批量插入方法

// 批量插入（忽略重复键）
int insertBatch(@Param("list") Collection<JavInfoDO> list);

对应的 XML（使用 MySQL INSERT IGNORE 或 ON DUPLICATE KEY UPDATE）：

<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>

JackettServiceImpl.dailyHot()：改造查重逻辑

// ❌ 旧：逐条 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());

JackettServiceImpl.dailyHot()：改造插入逻辑，循环结束后统一批量插入

// 循环内不再逐条 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 跑几分钟。

改动点：

JackettServiceImpl.java：引入并发控制

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

在类中声明线程池（或者使用 Spring 的 TaskExecutor）：

private static final ExecutorService SEARCH_EXECUTOR = Executors.newFixedThreadPool(5);

改造 dailyHot() 中的循环：

// ❌ 旧：串行 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() 同理也要改造。

JackettProperties.java：新增配置项

/** 搜索超时时间（分钟） */
private Integer searchTimeoutMinutes = 30;

application.yml：验证或添加配置
```
jackett:
  search-timeout-minutes: 30
```

RestTemplate 配置：在 JackettConfig 或 RestTemplate 的 @Bean 创建处，设置连接超时和读取超时

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，文件无限增长。

改动点：

JackettServiceImpl.dailyHot()：传入文件名时带上日期

// ❌ 旧
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);

或按年月日三级目录：

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");

或者如果决定改入库（推荐方案）：在 JavInfoDO 中新增 magnet_uri 字段，在 insertBatch 时一并写入，完全替代文件存储。

此时 dailyHot() 循环内去掉 saveFile 调用，toInsert 中设置 magnetUri。

注意事项：

如果选择入库替代文件，需要考虑 JavInfoDO 表结构变更和数据库 migration，改动量更大
如果继续用文件，建议增加清理策略（如保留最近 30 天的文件，自动删除过期文件）

P2-2：日期过滤改为时间窗口

描述：当前用 resultVO.getPublishDate().getYear() != currentYear 过滤，语义是「当年发布」而非「最近发布」。跨年时段逻辑偏差明显。

改动点：

JackettServiceImpl.dailyHot()：替换过滤条件

// ❌ 旧
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;
}

时间窗口大小建议外部化配置：

jackett:
  hot-window-hours: 24

P2-3：避免 Thread.sleep 阻塞 Quartz 线程

描述：Thread.sleep(searchDelay) 在 Quartz 的 @DisallowConcurrentExecution 工作线程中阻塞，无法处理其他 Job。

改动点：

如果改用了 P1-2 的并发方案（CompletableFuture），线程池已经独立于 Quartz 线程，sleep 问题自然解决。如果暂时不改并发，至少：

将 searchDelay 从 sleep 改为使用 ScheduledExecutorService 调度
或者在并发方案中就解决了，不用额外改动

推荐：P1-2 的并发方案已经涵盖了这一条，不需要独立处理。

P3-1：Redis Key 硬编码修复

描述：saveRedisIds() 中的 Redis key 硬编码为 "\""jav:rank:2023"\""，年份过时且带有多余引号。

改动点：

JackettProperties.java：新增配置项

/** Redis 热门排名 Key */
private String redisRankKey = "jav:rank";

JackettServiceImpl.saveRedisIds()：改为使用配置

// ❌ 旧
String key = "\""jav:rank:2023"\"";

// ✅ 新
String key = jackettProperties.getRedisRankKey();

application.yml：添加配置

jackett:
  redis-rank-key: "jav:rank:2026"  # 或更通用 key

P3-2：幂等性/中断支持

描述：Job 执行中断时，已 insert 的记录和已保存的文件产生不一致，下次执行产生重复数据。

改动点：

数据库层：JavInfoDO.jav_id 加 UNIQUE 约束，insertBatch 使用 INSERT IGNORE → 幂等写入，重复自动跳过
```
ALTER TABLE wellness_jav_info ADD UNIQUE INDEX uk_jav_id (jav_id);
```

文件层：按日期分片后（P2-1），同一天执行多次也是追加到同一个日期文件，仍然会重复。可以在写入前检查是否已存在：

// 方式 A：先读文件内容做 Set 去重（小文件可行，大文件低效）
// 方式 B：不管去重，反正按日期分片，运维定期清理即可

「执行中标记」：可选地在 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 执行两次，数据库和文件不产生重复记录

14 KiB Raw Blame History Unescape Escape

DailyHotJob 改造任务清单

改造目标

涉及文件清单

分步任务

P0-1：异常不再吞没，让 JobHandlerInvoker 能记录真实状态

P0-2：租户注解问题修正

P1-1：批量查重 + 批量插入替代 N+1

P1-2：并发搜索 + 超时控制

P2-1：文件按日期分片替代无限追加

P2-2：日期过滤改为时间窗口

P2-3：避免 Thread.sleep 阻塞 Quartz 线程

P3-1：Redis Key 硬编码修复

P3-2：幂等性/中断支持

验收条件

14 KiB

Raw Blame History