LeaderElection.java 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139
  1. package com.adx.tracking.leader;
  2. import com.adx.tracking.storage.RedisLock;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import java.time.Duration;
  6. import java.util.concurrent.Executors;
  7. import java.util.concurrent.ScheduledExecutorService;
  8. import java.util.concurrent.ScheduledFuture;
  9. import java.util.concurrent.TimeUnit;
  10. import java.util.concurrent.atomic.AtomicBoolean;
  11. import java.util.function.Consumer;
  12. /**
  13. * 对应 Go internal/leader/leader.go 的 Run()。
  14. * 通过 Redis 分布式锁竞选 Leader,赢得锁后执行 job 回调,
  15. * 同时启动后台线程定期续约(renewUntilDone),job 结束后释放锁。
  16. */
  17. public class LeaderElection {
  18. private static final Logger log = LoggerFactory.getLogger(LeaderElection.class);
  19. private final RedisLock redisLock;
  20. public LeaderElection(RedisLock redisLock) {
  21. this.redisLock = redisLock;
  22. }
  23. /**
  24. * 循环竞选,获得锁后执行 job(阻塞直到 job 返回或 stopped 为 true)。
  25. * 对应 Go leader.Run():job 通过 StopSignal 参数感知锁丢失(对应 Go 的 ctx.Done())。
  26. *
  27. * @param key Redis 锁 Key
  28. * @param ttl 锁 TTL
  29. * @param renewInterval 续约间隔
  30. * @param retryInterval 未获锁时重试间隔
  31. * @param stopped 外部停止信号(volatile boolean 引用)
  32. * @param job 获得锁后执行的任务,接收 per-election StopSignal(锁丢失时触发)
  33. * @param onError 错误回调
  34. */
  35. public void run(String key, Duration ttl, Duration renewInterval, Duration retryInterval,
  36. StopSignal stopped, Consumer<StopSignal> job, Consumer<Exception> onError) {
  37. Duration resolvedTtl = ttl.isZero() ? Duration.ofMinutes(2) : ttl;
  38. Duration resolvedRenewInterval = renewInterval.isZero() ? resolvedTtl.dividedBy(4) : renewInterval;
  39. Duration resolvedRetryInterval = retryInterval.isZero() ? Duration.ofSeconds(15) : retryInterval;
  40. while (!stopped.isStopped()) {
  41. Lease lease;
  42. try {
  43. log.info("[LeaderElection] trying to acquire lock: {}", key);
  44. lease = redisLock.tryAcquire(key, resolvedTtl);
  45. } catch (Exception e) {
  46. log.error("[LeaderElection] acquire lock {} failed: {}", key, e.getMessage(), e);
  47. report(onError, "acquire lock " + key + ": " + e.getMessage(), e);
  48. sleep(resolvedRetryInterval, stopped);
  49. continue;
  50. }
  51. if (lease == null) {
  52. // 未获到锁,等待后重试
  53. log.info("[LeaderElection] lock {} not acquired, retrying in {}ms", key, resolvedRetryInterval.toMillis());
  54. sleep(resolvedRetryInterval, stopped);
  55. continue;
  56. }
  57. log.info("[LeaderElection] lock {} acquired, starting job", key);
  58. // ─── 获得锁,创建 per-election 停止信号(对应 Go 的 jobCtx)────────
  59. // 必须用 AtomicBoolean 保证 scheduler 线程写入对 job 线程立即可见
  60. // (对应 Go context.Cancel 的天然 happens-before 语义)
  61. final AtomicBoolean renewFailed = new AtomicBoolean(false);
  62. StopSignal jobStop = () -> stopped.isStopped() || renewFailed.get();
  63. // ─── 启动续约线程(对应 Go renewUntilDone goroutine)────────────────
  64. ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
  65. Thread t = new Thread(r, "leader-renew-" + key);
  66. t.setDaemon(true);
  67. return t;
  68. });
  69. ScheduledFuture<?> renewFuture = scheduler.scheduleAtFixedRate(() -> {
  70. try {
  71. boolean ok = lease.renew(resolvedTtl);
  72. if (!ok) {
  73. report(onError, "lost lock " + key, null);
  74. renewFailed.set(true);
  75. }
  76. } catch (Exception e) {
  77. report(onError, "renew lock " + key + ": " + e.getMessage(), e);
  78. renewFailed.set(true);
  79. }
  80. }, resolvedRenewInterval.toMillis(), resolvedRenewInterval.toMillis(), TimeUnit.MILLISECONDS);
  81. // ─── 执行 job(传入 jobStop,job 应循环检查 jobStop.isStopped())────
  82. try {
  83. job.accept(jobStop);
  84. } catch (Exception e) {
  85. report(onError, "job error: " + e.getMessage(), e);
  86. } finally {
  87. renewFuture.cancel(true);
  88. scheduler.shutdownNow();
  89. try { lease.release(); } catch (Exception e) {
  90. report(onError, "release lock " + key + ": " + e.getMessage(), e);
  91. }
  92. }
  93. sleep(resolvedRetryInterval, stopped);
  94. }
  95. }
  96. private static void sleep(Duration d, StopSignal stopped) {
  97. long deadline = System.currentTimeMillis() + d.toMillis();
  98. while (!stopped.isStopped()) {
  99. long remaining = deadline - System.currentTimeMillis();
  100. if (remaining <= 0) break;
  101. try {
  102. Thread.sleep(Math.min(remaining, 200));
  103. } catch (InterruptedException e) {
  104. Thread.currentThread().interrupt();
  105. break;
  106. }
  107. }
  108. }
  109. private static void report(Consumer<Exception> onError, String msg, Exception cause) {
  110. if (onError != null) {
  111. onError.accept(cause != null ? cause : new RuntimeException(msg));
  112. } else {
  113. log.error(msg, cause);
  114. }
  115. }
  116. /** 外部停止信号接口(对应 Go 中的 ctx.Done()) */
  117. @FunctionalInterface
  118. public interface StopSignal {
  119. boolean isStopped();
  120. }
  121. }