订阅刷新改为每账号错峰,保证 24 小时内各刷一次

原先每 24 小时一次性刷新全部子账号:12 个账号在 16 秒内背靠背发完
24 个请求,且 @Scheduled(fixedRate) 未设 initialDelay,首次触发即立刻
执行,导致每次重启都重刷一遍全部账号——上游看到的请求密度实际取决于
部署频率。请求头也只带 HttpClient 默认值,上游收到的是
"Apache-HttpClient/5.6.4 (Java/25.0.4)"。

改为高频轻量 tick(5 分钟,延迟 2 分钟启动),只刷新已到期账号,单轮
最多 2 个。24 小时保证由三层构成:成功后按「成功时刻 + 窗口 − 一个 tick」
推进相位,使分散效果自我维持;到期时刻持久化在数据库,停机期间错过的
账号恢复后立即到期;兜底把被写到超过一个窗口之后的计划判为损坏并立即
刷新。失败用固定间隔重试(数据库没有连续失败次数字段,做不出真正的指数
退避),重试排在将来时被尊重,避免故障账号每个 tick 重打上游。从未成功过
的账号按 tick 间隔尽快刷,已有缓存的按窗口分散。超过两倍窗口仍未成功的
账号推送告警。

请求头按账号与订阅格式确定性地伪装成 mihomo/clash-verge/v2rayN 等真实
客户端身份;同一账号固定用同一身份,逐请求更换反而更像机器。实现前已
实测确认显式设置 Accept-Encoding 不会破坏 HttpClient5 的透明解压。注意
这只能伪装到 HTTP 头,TLS 指纹仍与真实客户端不同。

新增 next_refresh_at、last_success_epoch 两列,均为 Epoch 毫秒整数:SQLite
的 CURRENT_TIMESTAMP 写 UTC 文本而被 JDBC 按本地时区解释,实测偏差 8 小时,
不能用于时间比较。迁移为纯新增列,旧二进制忽略它们,故回滚程序无需回滚
数据库;回填 last_success_epoch 必须执行,否则既有账号会被当成从未成功过
而集中补刷,正好复现本次要消除的爆发。

同时移除 LocalService 的 24 小时全量入口与 refreshAll()(它们正是爆发式
写法),并将调度线程池由默认 1 调到 2,避免刷新阻塞连接自检。

测试 439 通过,指令覆盖率 82.1%。
This commit is contained in:
root
2026-09-15 16:15:41 +08:00
parent a0959618ca
commit 3adb42c538
23 changed files with 1776 additions and 199 deletions
@@ -2,6 +2,7 @@ package com.lion.lionwebsite.Configuration;
import com.lion.lionwebsite.Domain.*;
import com.lion.lionwebsite.Message.*;
import com.lion.lionwebsite.Service.SubscriptionRefreshPlanner;
import com.lion.lionwebsite.Util.GalleryUtil;
import com.pengrad.telegrambot.TelegramBot;
import com.pengrad.telegrambot.model.*;
@@ -14,6 +15,10 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.Clock;
import java.time.Duration;
import java.util.Random;
@Configuration
@RegisterReflectionForBinding(classes = {CustomConfiguration.class, GidToKey.class, ImageKeyCache.class,
GalleryForQuery.class, Gallery.class, GalleryTask.class, HikariConfig.class,
@@ -41,4 +46,22 @@ public class CustomBean {
public TelegramBot getTelegramBot(){
return new TelegramBot(botToken);
}
/**
* 订阅分散刷新的排程器。
*
* <p>放在这里是为了让窗口、最小间隔、tick 周期与重试间隔集中由配置注入,
* 同时保持 {@link SubscriptionRefreshPlanner} 本身是可直接构造的纯对象
* (便于用固定时钟与固定随机种子做确定性单元测试)。
*/
@Bean
public SubscriptionRefreshPlanner subscriptionRefreshPlanner(
@Value("${subscription.refresh.window-hours:24}") long windowHours,
@Value("${subscription.refresh.min-gap-minutes:60}") long minGapMinutes,
@Value("${subscription.refresh.tick-interval-ms:300000}") long tickIntervalMs,
@Value("${subscription.refresh.retry-delay-minutes:60}") long retryDelayMinutes) {
return new SubscriptionRefreshPlanner(Clock.systemDefaultZone(), new Random(),
Duration.ofHours(windowHours), Duration.ofMinutes(minGapMinutes),
Duration.ofMillis(tickIntervalMs), Duration.ofMinutes(retryDelayMinutes));
}
}
@@ -74,7 +74,7 @@ public class PersonalController {
@PostMapping("/updateSub")
public String updateSub() throws IOException {
Response response = Response.generateResponse();
if(localService.updateSub(true))
if(localService.updateSub())
response.success();
else
response.failure();
@@ -13,10 +13,10 @@ public interface SubMapper {
@Options(useGeneratedKeys = true, keyProperty = "id")
void insertSubscriptionAccount(SubscriptionAccount account);
@Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa order by id")
@Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, next_refresh_at as nextRefreshAt, last_success_epoch as lastSuccessEpoch, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa order by id")
ArrayList<SubscriptionAccount> selectAllSubscriptionAccounts();
@Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa where id=#{id}")
@Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, next_refresh_at as nextRefreshAt, last_success_epoch as lastSuccessEpoch, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa where id=#{id}")
SubscriptionAccount selectSubscriptionAccount(Integer id);
@Select("select count(*) from subscription_account where name=#{name}")
@@ -28,12 +28,18 @@ public interface SubMapper {
@Update("update subscription_account set name=#{name}, upstream_key=#{upstreamKey}, filter_high_multiplier=#{filterHighMultiplier}, enabled=#{enabled}, updated_at=CURRENT_TIMESTAMP, last_error=null where id=#{id}")
void updateSubscriptionAccount(SubscriptionAccount account);
@Update("update subscription_account set last_success_at=CURRENT_TIMESTAMP, last_error=null, updated_at=CURRENT_TIMESTAMP where id=#{id}")
void markSubscriptionRefreshSuccess(Integer id);
// last_success_at 仍写 SQLite 的 UTC 文本时间,仅供页面显示;
// last_success_epoch 是调度判断用的权威时刻,两者口径不同,不要互相推导。
@Update("update subscription_account set last_success_at=CURRENT_TIMESTAMP, last_success_epoch=#{epoch}, last_error=null, updated_at=CURRENT_TIMESTAMP where id=#{id}")
void markSubscriptionRefreshSuccess(@Param("id") Integer id, @Param("epoch") long epoch);
@Update("update subscription_account set last_error=#{error}, updated_at=CURRENT_TIMESTAMP where id=#{id}")
void markSubscriptionRefreshFailure(@Param("id") Integer id, @Param("error") String error);
/** 排定下一次刷新时刻;失败重试与次日分槽都走这里。 */
@Update("update subscription_account set next_refresh_at=#{nextRefreshAt}, updated_at=CURRENT_TIMESTAMP where id=#{id}")
void updateNextRefreshAt(@Param("id") Integer id, @Param("nextRefreshAt") long nextRefreshAt);
@Delete("delete from subscription_account where id=#{id}")
void deleteSubscriptionAccount(Integer id);
@@ -1,6 +1,5 @@
package com.lion.lionwebsite.Domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@@ -8,7 +7,6 @@ import java.util.Date;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class SubscriptionAccount {
private Integer id;
private String name;
@@ -22,4 +20,12 @@ public class SubscriptionAccount {
private Integer boundUserCount;
private String v2Url;
private String clashUrl;
/**
* 下一次应刷新的时刻(Epoch 毫秒),由分散调度器排定。
* 刻意用整数而非 DATETIME:SQLite 文本时间戳会被 JDBC 按本地时区解释,
* 实测偏差 8 小时,不能用于时间比较。
*/
private Long nextRefreshAt;
/** 最近一次成功刷新的真实时刻(Epoch 毫秒),用于「24 小时内必刷一次」的判断与陈旧告警。 */
private Long lastSuccessEpoch;
}
@@ -53,7 +53,7 @@ public class LocalService{
final RemoteService remoteService;
final SubscriptionRefreshService subscriptionRefreshService;
final SubscriptionRefreshScheduler subscriptionRefreshScheduler;
/**
* 检查连接是否有效,如果无效自动重连
@@ -104,115 +104,23 @@ public class LocalService{
}
/**
* 定时更新订阅
* @throws IOException 下载以及保存异常
* 手动「更新订阅」:立即刷新全部启用账号,供管理页按钮调用。
*
* <p>定时刷新已移交 {@link SubscriptionRefreshScheduler}。原先这里还有一个
* {@code @Scheduled(fixedRate = 86400000)} 的定时入口,它会一次性刷新全部账号
* (十几个账号在十几秒内打满),且因为没有 initialDelay,每次重启都会立刻重刷一遍,
* 使上游看到的请求密度取决于部署频率。该入口已移除。
*
* <p>不再保留 {@code isManual} 参数:定时路径已不存在,留着它只会让人以为
* 「传 false 就是不刷新」,而实际语义是「什么都不做却报成功」。
*/
@Scheduled(fixedRate = 86400000)
public void updateSubScheduler() throws IOException {
updateSub(false);
}
/**
* 更新订阅链接的实际方法
*/
public boolean updateSub(boolean isManual) throws IOException {
// 手动和定时入口均刷新全部启用子账号;全部成功后更新原有“上次更新时间”。
boolean success = subscriptionRefreshService.refreshAll();
public boolean updateSub() {
boolean success = subscriptionRefreshScheduler.refreshAllNow();
remoteService.requestSubscriptionSync();
if (success)
configurationMapper.updateConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME,
CustomUtil.dateTimeFormatter().format(LocalDateTime.now()));
return success;
/*
DateTimeFormatter dateTimeFormatter = CustomUtil.dateTimeFormatter();
CustomConfiguration customConfiguration = configurationMapper.selectConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME);
//如果不是手动,则判断更新间隔是否满足,不满足则取消更新
if (!isManual) {
LocalDateTime lastUpdate = LocalDateTime.parse(customConfiguration.getValue(), CustomUtil.dateTimeFormatter());
LocalDateTime now = LocalDateTime.now();
now = now.plusHours(-3);
if(now.isBefore(lastUpdate))
return false;
}
File DouNaiClashFile = new File("sub/DouNaiClash.txt");
File DouNaiV2rayFile = new File("sub/DouNaiV2ray.txt");
File directory = new File("sub");
if(!directory.isDirectory())
Files.createDirectory(Paths.get("sub"));
List<String> DouNaiClash_profile;
//下载豆奶v2ray订阅
try(FileWriter writer = new FileWriter(DouNaiV2rayFile)) {
String DouNaiV2rayRaw = Get(DouNaiV2ray).getFirst();
String[] v2rayPlain = new String(Base64.getDecoder().decode(DouNaiV2rayRaw)).split("\n");
StringBuilder stringBuilder = new StringBuilder();
Pattern pattern = Pattern.compile("-?\\d+(\\.\\d+)?");
//过滤高倍率节点
for(String node: v2rayPlain){
String name = URLDecoder.decode(node.split("#")[1], StandardCharsets.UTF_8);
if(name.contains("流量")){
Matcher matcher = pattern.matcher(name.substring(name.indexOf("(") + 1, name.indexOf(")")));
if (matcher.find()) {
// 将匹配到的数字添加到列表中
float ratio = Float.parseFloat(matcher.group());
if(ratio <= 2) {
stringBuilder.append(node).append("\n");
continue;
}
}
stringBuilder.append(node).append("\n");
}
else{
stringBuilder.append(node).append("\n");
}
}
writer.write(new String(Base64.getEncoder().encode(stringBuilder.toString().getBytes(StandardCharsets.UTF_8))));
log.info("load DouNai v2ray complete");
}catch (IOException e){
log.error("load DouNai v2ray failure", e);
}
//下载豆奶clash订阅
try(FileWriter writer = new FileWriter(DouNaiClashFile)) {
DouNaiClash_profile = Get(DouNaiClash);
//过滤高倍率节点
ArrayList<String> clashProcessed = new ArrayList<>();
boolean isProxies = false;
boolean skip = false;
for(String line: DouNaiClash_profile){
if(line.equals("proxies:"))
isProxies = true;
else if(line.equals("proxy-groups:") && isProxies)
isProxies = false;
if(isProxies) {
if (line.contains("name"))
skip = line.contains("流量");
if (!skip)
clashProcessed.add(line);
}
else
if (!line.contains("流量"))
clashProcessed.add(line);
}
for(String line: clashProcessed)
writer.write(line + "\n");
log.info("load DouNai clash complete");
}catch (IOException e){
log.error("load DouNai clash failure", e);
}
configurationMapper.updateConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME, dateTimeFormatter.format(LocalDateTime.now()));
return true;
*/
}
/**
@@ -46,7 +46,11 @@ public class SubService {
return response.failure("名称和上游 key 不能为空").toJSONString();
if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0)
return response.failure("名称或上游 key 已存在").toJSONString();
account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null);
account = new SubscriptionAccount();
account.setName(name.trim());
account.setUpstreamKey(upstreamKey.trim());
account.setFilterHighMultiplier(filterHighMultiplier);
account.setEnabled(enabled);
subMapper.insertSubscriptionAccount(account);
} finally {
lock.unlock();
@@ -0,0 +1,225 @@
package com.lion.lionwebsite.Service;
import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
/**
* 为一批子账号排定刷新时刻。
*
* <p>目标:每个账号每 24 小时内至少刷新一次,且各账号的时刻互相错开,
* 不出现「一次全量、十几个账号在十几秒内打完」的爆发式请求。
*
* <p>排程模型是「错峰 + 固定相位」:
* <ol>
* <li><b>错峰</b>:首次排程把账号随机铺满一个窗口,两两之间至少隔开 {@code minGap};</li>
* <li><b>固定相位</b>:账号每次成功后按窗口长度(略减一个 tick)推进,
* 因此各自的相位保持不变,分散效果自我维持,不需要每天重新洗牌;</li>
* <li><b>兜底</b>:任何情况下下一次刷新都不会晚于「上次成功 + 窗口」,
* 把「24 小时内必然刷新一次」从概率保证变成确定保证。</li>
* </ol>
*
* <p>本类是纯函数式设计(注入 {@link Clock} 与带种子的 {@link Random}),
* 不触碰数据库与网络,因此可以直接做确定性单元测试。
*/
public class SubscriptionRefreshPlanner {
private final Clock clock;
private final Random random;
private final Duration window;
private final Duration minGap;
/** 一个调度周期的最大延迟;用它给窗口留余量,使实际间隔严格落在窗口内。 */
private final Duration tickInterval;
/** 刷新失败后的重试间隔。 */
private final Duration retryDelay;
/**
* @param clock 时间源,测试可注入固定时钟
* @param random 随机源,测试可注入固定种子以复现排程
* @param window 刷新窗口,即「24 小时内至少一次」中的 24 小时
* @param minGap 相邻两个账号之间的最小间隔
* @param tickInterval 调度周期,用于给窗口留出余量
* @param retryDelay 失败后的重试间隔
*/
public SubscriptionRefreshPlanner(Clock clock, Random random, Duration window, Duration minGap,
Duration tickInterval, Duration retryDelay) {
if (window.isZero() || window.isNegative())
throw new IllegalArgumentException("刷新窗口必须为正数");
if (minGap.isNegative())
throw new IllegalArgumentException("最小间隔不能为负数");
if (tickInterval.isNegative())
throw new IllegalArgumentException("调度周期不能为负数");
if (retryDelay.isZero() || retryDelay.isNegative())
throw new IllegalArgumentException("重试间隔必须为正数");
if (retryDelay.compareTo(window) > 0)
throw new IllegalArgumentException("重试间隔不能大于刷新窗口,否则无法保证窗口内重试");
this.clock = clock;
this.random = random;
this.window = window;
this.minGap = minGap;
this.tickInterval = tickInterval;
this.retryDelay = retryDelay;
}
/** 便捷构造:重试间隔默认取 1 小时。 */
public SubscriptionRefreshPlanner(Clock clock, Random random, Duration window, Duration minGap,
Duration tickInterval) {
this(clock, random, window, minGap, tickInterval, Duration.ofHours(1));
}
/**
* 账号数与最小间隔是否放得下。
*
* <p>注意这里刻意只用于「提示」而不是阻断启动:账号数是由用户在管理页决定的,
* 若因为它超限就让整个应用起不来,等于把配置细节变成一次线上故障。
* 放不下时 {@link #initialSchedule} 仍会把账号铺满整个窗口(只是间隔小于期望值),
* 结果是错峰效果变弱,而不是退回串行爆发。
*/
public boolean isCapacitySufficient(int accountCount) {
if (accountCount <= 1)
return true;
return minGap.toMillis() * (accountCount - 1) <= window.toMillis();
}
/** 容量不足时的告警文案,供调用方记录日志。 */
public String capacityMessage(int accountCount) {
long required = minGap.toMillis() * (accountCount - 1);
return String.format(
"刷新窗口放不下最小间隔:账号 %d 个、最小间隔 %d 分钟,需要 %d 分钟,但窗口只有 %d 分钟;"
+ "实际仍会铺满窗口,只是账号间间隔会小于期望值。可调小 subscription.refresh.min-gap-minutes。",
accountCount, minGap.toMinutes(),
Duration.ofMillis(required).toMinutes(), window.toMinutes());
}
/**
* 为一批「已成功刷新过」的账号铺开计划,返回顺序与 accountIds 一致。
*
* <p>已成功过的账号内容仍然可用,因此可以安心分散到整个窗口,
* 不会因为排在 20 小时后而让用户拿不到订阅。
*/
public List<Long> initialSchedule(List<Integer> accountIds) {
List<Long> result = new ArrayList<>(accountIds.size());
if (accountIds.isEmpty())
return result;
long now = clock.millis();
long windowMillis = window.toMillis();
int count = accountIds.size();
long slotWidth = windowMillis / count;
long jitterCap = Math.max(1, Math.min(slotWidth, Math.max(1, slotWidth - minGap.toMillis())));
List<Integer> slots = new ArrayList<>(count);
for (int i = 0; i < count; i++)
slots.add(i);
Collections.shuffle(slots, random);
long[] planned = new long[count];
for (int accountIndex = 0; accountIndex < count; accountIndex++) {
long offset = slots.get(accountIndex) * slotWidth + (long) (random.nextDouble() * jitterCap);
planned[accountIndex] = now + Math.min(offset, windowMillis - 1);
}
for (int i = 0; i < count; i++)
result.add(planned[i]);
return result;
}
/**
* 为尚未成功过的账号排程:尽快刷,但互相错开。
*
* <p>这类账号(新建、或上游 key 刚改过)可能根本没有可用缓存,用户立刻请求订阅
* 就会拿不到内容,因此不能像常规账号那样分散到整个窗口。这里把它们按
* {@code spacing} 依次排开,配合调度器的单轮上限逐批刷新,
* 既尽早补齐,又避免一批新账号同时打上游。
*
* @param accountIds 账号 ID,顺序即刷新优先级
* @param spacing 相邻两个账号的间隔(通常取一个 tick)
*/
public List<Long> coldStartSchedule(List<Integer> accountIds, Duration spacing) {
List<Long> result = new ArrayList<>(accountIds.size());
long now = clock.millis();
long step = Math.max(1, spacing.toMillis());
for (int i = 0; i < accountIds.size(); i++)
result.add(now + i * step);
return result;
}
/**
* 已排定的时刻是否已到期。
*
* <p>直接以排定时刻为准,这样分散效果与失败退避都得以保持。
* 只额外兜住一种异常:计划被写到超过一个窗口之后。这不可能是本调度器产生的
* 排程(成功时排在窗口减一个 tick 之后,失败时排在 4 小时以内),
* 按「计划损坏」处理并立即到期,从而保证 24 小时内必定刷新。
*
* <p>尚未排程({@code scheduled == null})视为立即到期,由调用方先行铺开计划。
*/
public boolean isDue(Long scheduled, long now) {
if (scheduled == null)
return true;
if (scheduled - now > window.toMillis())
return true;
return scheduled <= now;
}
/**
* 成功之后的下一次刷新时刻。
*
* <p>只推进「窗口 - 一个调度周期」,使 tick 即使晚触发也不会突破 24 小时;
* 从上次成功的时刻起算,保证相邻两次的间隔严格小于窗口。
*/
public long nextAfterSuccess(long successMillis) {
long advance = Math.max(1, window.minus(tickInterval).toMillis());
return successMillis + advance;
}
/**
* 失败之后的重试时刻。
*
* <p>用固定的重试间隔而不是「失败次数 × 指数退避」:刷新成功会清空 {@code last_error},
* 数据库里没有「连续失败次数」这一栏,退避指数实际上只能取到 0 或 1,
* 写出来会让人误以为有指数退避而其实没有。固定的重试间隔行为可预期,
* 且远小于窗口,不会威胁「24 小时内尝试刷新一次」。
*
* <p>唯一需要裁剪的情况:若「上次成功 + 窗口」还没到,则重试不得晚于它;
* 已经越过该点时保留重试时刻本身——重复请求并不会提高成功率,
* 反过来在每个 tick 重打上游才是真正的问题,这种情况交由陈旧告警处理。
*
* @param now 当前时刻
* @param lastSuccess 上次成功时刻;未知时传 null,不做裁剪
*/
public long nextAfterFailure(long now, Long lastSuccess) {
long candidate = now + retryDelay.toMillis();
if (lastSuccess == null)
return candidate;
long deadline = lastSuccess + window.toMillis();
return deadline > now ? Math.min(candidate, deadline) : candidate;
}
/** 是否已越过兜底线(距上次成功已达一个窗口),用于诊断与告警判定。 */
public boolean isOverdue(Long lastSuccess, long now) {
return lastSuccess != null && now - lastSuccess >= window.toMillis();
}
private long now() {
return clock.millis();
}
public Clock clock() {
return clock;
}
public Duration window() {
return window;
}
public Duration minGap() {
return minGap;
}
public Duration tickInterval() {
return tickInterval;
}
}
@@ -0,0 +1,240 @@
package com.lion.lionwebsite.Service;
import com.lion.lionwebsite.Dao.normal.SubMapper;
import com.lion.lionwebsite.Domain.SubscriptionAccount;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* 分散刷新各子账号的订阅。
*
* <p>取代原先「每 24 小时一次全量」的调度。旧写法有两个问题:所有账号在十几秒内
* 背靠背打完,且 {@code @Scheduled(fixedRate = 86400000)} 没有 initialDelay,
* 首次触发即立刻执行——于是每次重启都会重刷一遍全部账号,上游看到的请求密度取决于
* 部署频率,而不是每天一次。
*
* <p>现在改为高频轻量 tick,每轮只刷新已经到期的账号。「24 小时内每个账号至少
* 更新一次」由三层保证:
* <ol>
* <li>成功后排定「成功时刻 + 窗口 - 一个 tick」,相位自我维持,分散效果不会因
* 重启或重排而退化;</li>
* <li>到期时刻存在数据库里,停机期间错过计划的账号在恢复后立即到期;</li>
* <li>兜底闸:任何排定时刻都被裁剪到「上次成功 + 窗口」以内,越线即立刻刷新。</li>
* </ol>
*/
@Service
@Slf4j
public class SubscriptionRefreshScheduler {
private final SubMapper subMapper;
private final SubscriptionRefreshService refreshService;
private final SubscriptionRefreshPlanner planner;
private final PushService pushService;
/** 单个 tick 最多刷新几个账号,避免重启后一批到期账号同时开火。 */
@Value("${subscription.refresh.max-per-tick:2}")
int maxPerTick;
/** 超过窗口的这个倍数仍未成功即告警,避免静默失败。 */
@Value("${subscription.refresh.stale-alert-multiplier:2.0}")
double staleAlertMultiplier;
/** 已告警过的账号,避免每轮重复推送;刷新成功后移除。 */
private final Set<Integer> alertedAccounts = new HashSet<>();
public SubscriptionRefreshScheduler(SubMapper subMapper,
SubscriptionRefreshService refreshService,
SubscriptionRefreshPlanner planner,
PushService pushService) {
this.subMapper = subMapper;
this.refreshService = refreshService;
this.planner = planner;
this.pushService = pushService;
}
/**
* 轻量 tick:只处理已到期的账号,且单轮有数量上限。
*
* <p>用 fixedDelay 而非 fixedRate,一轮结束再计下一轮,避免上一轮没跑完就叠加触发;
* initialDelay 让重启后的第一轮稍晚开始,先让应用完成启动。
*/
@Scheduled(fixedDelayString = "${subscription.refresh.tick-interval-ms:300000}",
initialDelayString = "${subscription.refresh.initial-delay-ms:120000}")
public void tick() {
try {
refreshDueAccounts();
} catch (Exception e) {
// 调度方法抛出异常会导致后续触发被取消,这里必须兜住。
log.error("订阅分散刷新 tick 失败", e);
}
}
/**
* 刷新当前已到期的账号。
*
* @return 本轮实际刷新的账号数
*/
public int refreshDueAccounts() {
List<SubscriptionAccount> enabled = enabledAccounts();
ensureScheduled(enabled);
long now = planner.clock().millis();
int refreshed = 0;
for (SubscriptionAccount account : enabled) {
if (refreshed >= maxPerTick)
break;
if (!isDue(account, now))
continue;
refreshOne(account, now);
refreshed++;
}
reportStale(enabled, now);
return refreshed;
}
/**
* 手动刷新入口:立即刷新全部启用账号。
*
* <p>刻意不受 {@code maxPerTick} 限制——管理页「更新订阅」的语义就是立刻全部刷新。
* 但成功后同样要重排各自的下一次时刻,否则刚手动刷完会让当天的计划作废,
* 该账号反而可能超过 24 小时没有下次更新。
*/
public boolean refreshAllNow() {
boolean success = true;
long now = planner.clock().millis();
for (SubscriptionAccount account : enabledAccounts()) {
if (!refreshOne(account, now))
success = false;
}
return success;
}
private List<SubscriptionAccount> enabledAccounts() {
return subMapper.selectAllSubscriptionAccounts().stream()
.filter(SubscriptionAccount::isEnabled)
.sorted(Comparator.comparing(SubscriptionAccount::getId))
.toList();
}
/** 刷新单个账号并按结果重排下一次时刻。 */
private boolean refreshOne(SubscriptionAccount account, long now) {
boolean ok = refreshService.refresh(account.getId());
if (ok) {
long next = planner.nextAfterSuccess(planner.clock().millis());
subMapper.updateNextRefreshAt(account.getId(), next);
alertedAccounts.remove(account.getId());
log.info("订阅刷新成功 accountId={} 下次刷新={}", account.getId(), next);
} else {
long next = planner.nextAfterFailure(now, account.getLastSuccessEpoch());
subMapper.updateNextRefreshAt(account.getId(), next);
log.warn("订阅刷新失败 accountId={} 下次重试={}", account.getId(), next);
}
return ok;
}
/**
* 账号是否到期。
*
* <p>以排定时刻为准;尚未排程或计划明显损坏(超出窗口)时视为立即到期,
* 由 {@link #ensureScheduled} 先铺开或纠正。
*/
private boolean isDue(SubscriptionAccount account, long now) {
return planner.isDue(account.getNextRefreshAt(), now);
}
/**
* 为尚无计划的账号补齐计划。
*
* <p>按紧迫性分三类,因为它们对「排到多晚」的容忍度不同:
* <ul>
* <li><b>从未成功过</b>(新建账号、刚改过上游 key):可能还没有可用缓存,
* 用户此时请求订阅会拿不到内容,因此尽快刷,只按 tick 间隔互相错开;</li>
* <li><b>已成功过但已超期</b>(距上次成功已达一个窗口):保证即将失效,
* 必须立即刷新,不能排到窗口内更晚的位置;</li>
* <li><b>已成功过且未超期</b>:缓存仍可分发,按窗口分散即可,不必挤在一起。</li>
* </ul>
*/
private void ensureScheduled(List<SubscriptionAccount> enabled) {
long now = planner.clock().millis();
List<SubscriptionAccount> cold = new ArrayList<>();
List<SubscriptionAccount> overdue = new ArrayList<>();
List<SubscriptionAccount> warm = new ArrayList<>();
for (SubscriptionAccount account : enabled) {
if (account.getNextRefreshAt() != null)
continue;
if (account.getLastSuccessEpoch() == null)
cold.add(account);
else if (planner.isOverdue(account.getLastSuccessEpoch(), now))
overdue.add(account);
else
warm.add(account);
}
if (cold.isEmpty() && overdue.isEmpty() && warm.isEmpty())
return;
if (!planner.isCapacitySufficient(enabled.size()))
log.warn("{}", planner.capacityMessage(enabled.size()));
// 超期账号立即到期(计划设为当前时刻),本轮就会被刷新。
for (SubscriptionAccount account : overdue)
schedule(account, now);
int index = 0;
List<Long> coldPlan = planner.coldStartSchedule(
cold.stream().map(SubscriptionAccount::getId).toList(), planner.tickInterval());
for (SubscriptionAccount account : cold)
schedule(account, coldPlan.get(index++));
List<Integer> warmIds = warm.stream().map(SubscriptionAccount::getId).toList();
List<Long> warmPlan = planner.initialSchedule(warmIds);
for (int i = 0; i < warm.size(); i++)
schedule(warm.get(i), warmPlan.get(i));
}
/**
* 写入新计划,并同步更新内存中的对象。
*
* <p>必须同时更新内存副本:否则刚被排到几小时后的账号在本轮仍带着
* {@code nextRefreshAt == null},会被 {@link #isDue} 判为立即到期并当场刷新,
* 于是「铺开」在第一个 tick 就失效了(迁移后首批账号会全部集中刷新)。
*/
private void schedule(SubscriptionAccount account, long nextRefreshAt) {
subMapper.updateNextRefreshAt(account.getId(), nextRefreshAt);
account.setNextRefreshAt(nextRefreshAt);
}
/** 陈旧账号告警:同一账号在连续陈旧期间只推一次,刷新成功后重置。 */
private void reportStale(List<SubscriptionAccount> enabled, long now) {
long threshold = (long) (planner.window().toMillis() * staleAlertMultiplier);
for (SubscriptionAccount account : enabled) {
Long lastSuccess = account.getLastSuccessEpoch();
if (lastSuccess == null || now - lastSuccess < threshold)
continue;
if (!alertedAccounts.add(account.getId()))
continue;
long hours = (now - lastSuccess) / 3_600_000L;
String message = String.format("订阅账号 %d(%s)已 %d 小时未成功刷新,最近错误:%s",
account.getId(), account.getName(), hours,
account.getLastError() == null ? "无" : account.getLastError());
log.error("{}", message);
pushService.sendToMe(message);
}
// 已删除或停用的账号不再保留告警状态,避免集合无限增长。
Set<Integer> alive = new HashSet<>();
for (SubscriptionAccount account : enabled)
alive.add(account.getId());
alertedAccounts.retainAll(alive);
}
SubscriptionRefreshPlanner planner() {
return planner;
}
}
@@ -2,6 +2,7 @@ package com.lion.lionwebsite.Service;
import com.lion.lionwebsite.Dao.normal.SubMapper;
import com.lion.lionwebsite.Domain.SubscriptionAccount;
import com.lion.lionwebsite.Util.SubscriptionClientProfile;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hc.client5.http.classic.methods.HttpGet;
@@ -18,6 +19,7 @@ import java.io.*;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.time.Clock;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -52,14 +54,8 @@ public class SubscriptionRefreshService {
@Value("${subscription.cache-root:sub/accounts}")
String cacheRoot;
public boolean refreshAll() {
boolean success = true;
for (SubscriptionAccount account : subMapper.selectAllSubscriptionAccounts()) {
if (account.isEnabled() && !refresh(account.getId()))
success = false;
}
return success;
}
/** 时间源,测试可替换以固定成功时刻。 */
Clock clock = Clock.systemDefaultZone();
private long refreshSequence;
// Accessed only while holding the coordinator write lock.
@@ -82,8 +78,10 @@ public class SubscriptionRefreshService {
// Network access and parsing never hold the shared subscription lock.
try {
String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold);
List<String> clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold);
String v2 = processV2(firstLine(download(v2Url(account), SubscriptionClientProfile.forAccount(accountId, false))),
account.isFilterHighMultiplier(), highMultiplierThreshold);
List<String> clash = processClash(download(clashUrl(account), SubscriptionClientProfile.forAccount(accountId, true)),
account.isFilterHighMultiplier(), highMultiplierThreshold);
stateLock.lock();
try {
if (!isCurrent(account, version))
@@ -92,7 +90,7 @@ public class SubscriptionRefreshService {
Files.createDirectories(dir);
atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8));
atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8));
subMapper.markSubscriptionRefreshSuccess(accountId);
subMapper.markSubscriptionRefreshSuccess(accountId, clock.millis());
return true;
} finally {
stateLock.unlock();
@@ -219,8 +217,24 @@ public class SubscriptionRefreshService {
return matcher.find() && Double.parseDouble(matcher.group(1)) > threshold;
}
List<String> download(String url) throws IOException {
/**
* 下载上游订阅正文。
*
* @param profile 要伪装的客户端身份;为 null 时保留 HttpClient 默认头(仅测试使用)。
* 生产路径必须传入,否则上游会看到 {@code Apache-HttpClient/... (Java/...)}。
*/
List<String> download(String url, SubscriptionClientProfile profile) throws IOException {
HttpGet get = new HttpGet(url);
if (profile != null) {
get.addHeader("User-Agent", profile.userAgent());
get.addHeader("Accept", profile.accept());
get.addHeader("Accept-Language", profile.acceptLanguage());
// 实测:显式设置 Accept-Encoding 不会破坏 HttpClient5 的透明解压,
// ContentCompressionExec 仍按 Content-Encoding 正确解码。
get.addHeader("Accept-Encoding", "gzip, deflate");
// 订阅客户端是长连接复用,且不会发送 Referer。
get.addHeader("Connection", "keep-alive");
}
try (CloseableHttpResponse response = HTTP_CLIENT.execute(get)) {
if (response.getCode() != 200)
throw new IOException("上游 HTTP 状态码 " + response.getCode());
@@ -0,0 +1,65 @@
package com.lion.lionwebsite.Util;
import java.util.List;
/**
* 抓取上游订阅时伪装的客户端身份。
*
* <p>此前请求只带 HttpClient 的默认头,实测上游收到的是
* {@code User-agent: Apache-HttpClient/5.6.4 (Java/25.0.4)},
* 对订阅站点而言等同于「这是一个 Java 程序在批量拉取」。
* 这里按订阅格式给出与真实客户端一致的常见头组合。
*
* <p>注意:只能伪装到 HTTP 头这一层。真实 Clash/Mihomo 使用 Go 的 TLS 栈与
* HTTP/2,TLS 指纹与 Java HttpClient 仍有差异;若上游做主动指纹识别,
* 需要另外引入真实客户端进程,本枚举解决不了。
*/
public enum SubscriptionClientProfile {
MIHOMO("mihomo/v1.19.2", "application/yaml, */*", "zh-CN,zh;q=0.9"),
CLASH_VERGE("clash-verge/v2.0.3", "application/yaml, */*", "zh-CN,zh;q=0.9"),
CLASH_WINDOWS("ClashforWindows/0.20.39", "*/*", "zh-CN,zh;q=0.9"),
V2RAYN("v2rayN/6.45", "*/*", "zh-CN,zh;q=0.9"),
V2RAYNG("v2rayNG/1.9.16", "*/*", "zh-CN,zh;q=0.9"),
SHADOWROCKET("Shadowrocket/2.2.39", "*/*", "zh-CN,zh;q=0.9");
static final List<SubscriptionClientProfile> CLASH_PROFILES = List.of(MIHOMO, CLASH_VERGE, CLASH_WINDOWS);
static final List<SubscriptionClientProfile> V2_PROFILES = List.of(V2RAYN, V2RAYNG, SHADOWROCKET);
private final String userAgent;
private final String accept;
private final String acceptLanguage;
SubscriptionClientProfile(String userAgent, String accept, String acceptLanguage) {
this.userAgent = userAgent;
this.accept = accept;
this.acceptLanguage = acceptLanguage;
}
public String userAgent() {
return userAgent;
}
public String accept() {
return accept;
}
public String acceptLanguage() {
return acceptLanguage;
}
/**
* 按账号与订阅格式确定性地选择一个身份。
*
* <p>刻意不使用随机数:同一个账号每次都用同一个客户端,才符合真实用户的样子;
* 逐请求更换 User-Agent 本身就是更明显的机器特征。不同账号使用不同客户端,
* 又便于在必要时按账号区分上游流量。
*
* @param accountId 子账号 ID,可为 null(视为 0)
* @param clash true 表示 Clash 格式,false 表示 V2Ray 格式
*/
public static SubscriptionClientProfile forAccount(Integer accountId, boolean clash) {
List<SubscriptionClientProfile> candidates = clash ? CLASH_PROFILES : V2_PROFILES;
int id = accountId == null ? 0 : accountId;
return candidates.get(Math.floorMod(id, candidates.size()));
}
}
+19
View File
@@ -12,6 +12,12 @@ spring:
datasource-cache:
driver-class-name: org.sqlite.JDBC
jdbc-url: jdbc:sqlite:cache.db
# 定时任务默认只有 1 个线程。订阅刷新一轮最多 2 个账号、每个 2 次请求(各 15 秒超时),
# 最坏可占用约 1 分钟;单线程会让连接自检、cookie 检测等任务在此期间全部排队。
task:
scheduling:
pool:
size: 2
mvc:
view:
prefix: /resources/templates/
@@ -38,6 +44,19 @@ local:
subscription:
cache-root: sub/accounts
# 分散刷新:每个账号在 window-hours 内至少刷新一次,两两之间至少相隔 min-gap-minutes;
# max-per-tick 限制单个 tick 刷新几个账号,避免重启后一批到期账号同时开火。
refresh:
window-hours: 24
min-gap-minutes: 60
# 5 分钟一个轻量 tick,只处理已到期账号。
tick-interval-ms: 300000
# 重启后延迟 2 分钟再开始,先让应用完成启动。
initial-delay-ms: 120000
max-per-tick: 2
# 失败后的重试间隔,必须小于 window-hours。
retry-delay-minutes: 60
stale-alert-multiplier: 2.0
upstream:
# Use environment variables or an external config file in production.
v2-url-template: "https://aaaa.gay/link/{key}?client=v2"