package com.adx.tracking.leader; import com.adx.tracking.storage.RedisLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; /** * 对应 Go internal/leader/leader.go 的 Run()。 * 通过 Redis 分布式锁竞选 Leader,赢得锁后执行 job 回调, * 同时启动后台线程定期续约(renewUntilDone),job 结束后释放锁。 */ public class LeaderElection { private static final Logger log = LoggerFactory.getLogger(LeaderElection.class); private final RedisLock redisLock; public LeaderElection(RedisLock redisLock) { this.redisLock = redisLock; } /** * 循环竞选,获得锁后执行 job(阻塞直到 job 返回或 stopped 为 true)。 * 对应 Go leader.Run():job 通过 StopSignal 参数感知锁丢失(对应 Go 的 ctx.Done())。 * * @param key Redis 锁 Key * @param ttl 锁 TTL * @param renewInterval 续约间隔 * @param retryInterval 未获锁时重试间隔 * @param stopped 外部停止信号(volatile boolean 引用) * @param job 获得锁后执行的任务,接收 per-election StopSignal(锁丢失时触发) * @param onError 错误回调 */ public void run(String key, Duration ttl, Duration renewInterval, Duration retryInterval, StopSignal stopped, Consumer job, Consumer onError) { Duration resolvedTtl = ttl.isZero() ? Duration.ofMinutes(2) : ttl; Duration resolvedRenewInterval = renewInterval.isZero() ? resolvedTtl.dividedBy(4) : renewInterval; Duration resolvedRetryInterval = retryInterval.isZero() ? Duration.ofSeconds(15) : retryInterval; while (!stopped.isStopped()) { Lease lease; try { log.info("[LeaderElection] trying to acquire lock: {}", key); lease = redisLock.tryAcquire(key, resolvedTtl); } catch (Exception e) { log.error("[LeaderElection] acquire lock {} failed: {}", key, e.getMessage(), e); report(onError, "acquire lock " + key + ": " + e.getMessage(), e); sleep(resolvedRetryInterval, stopped); continue; } if (lease == null) { // 未获到锁,等待后重试 log.info("[LeaderElection] lock {} not acquired, retrying in {}ms", key, resolvedRetryInterval.toMillis()); sleep(resolvedRetryInterval, stopped); continue; } log.info("[LeaderElection] lock {} acquired, starting job", key); // ─── 获得锁,创建 per-election 停止信号(对应 Go 的 jobCtx)──────── // 必须用 AtomicBoolean 保证 scheduler 线程写入对 job 线程立即可见 // (对应 Go context.Cancel 的天然 happens-before 语义) final AtomicBoolean renewFailed = new AtomicBoolean(false); StopSignal jobStop = () -> stopped.isStopped() || renewFailed.get(); // ─── 启动续约线程(对应 Go renewUntilDone goroutine)──────────────── ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "leader-renew-" + key); t.setDaemon(true); return t; }); ScheduledFuture renewFuture = scheduler.scheduleAtFixedRate(() -> { try { boolean ok = lease.renew(resolvedTtl); if (!ok) { report(onError, "lost lock " + key, null); renewFailed.set(true); } } catch (Exception e) { report(onError, "renew lock " + key + ": " + e.getMessage(), e); renewFailed.set(true); } }, resolvedRenewInterval.toMillis(), resolvedRenewInterval.toMillis(), TimeUnit.MILLISECONDS); // ─── 执行 job(传入 jobStop,job 应循环检查 jobStop.isStopped())──── try { job.accept(jobStop); } catch (Exception e) { report(onError, "job error: " + e.getMessage(), e); } finally { renewFuture.cancel(true); scheduler.shutdownNow(); try { lease.release(); } catch (Exception e) { report(onError, "release lock " + key + ": " + e.getMessage(), e); } } sleep(resolvedRetryInterval, stopped); } } private static void sleep(Duration d, StopSignal stopped) { long deadline = System.currentTimeMillis() + d.toMillis(); while (!stopped.isStopped()) { long remaining = deadline - System.currentTimeMillis(); if (remaining <= 0) break; try { Thread.sleep(Math.min(remaining, 200)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } private static void report(Consumer onError, String msg, Exception cause) { if (onError != null) { onError.accept(cause != null ? cause : new RuntimeException(msg)); } else { log.error(msg, cause); } } /** 外部停止信号接口(对应 Go 中的 ctx.Done()) */ @FunctionalInterface public interface StopSignal { boolean isStopped(); } }