| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139 |
- 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<StopSignal> job, Consumer<Exception> 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<Exception> 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();
- }
- }
|