修改报错

src/main/java/com/adx/tencent/AppConfiguration.java

@@ -51,7 +51,6 @@ import com.adx.tencent.oppo.store.OppoHotStore;
 import com.adx.tencent.oppo.service.OppoTagEventResolver;
 import com.adx.tencent.oppo.service.OppoAuthService;
 import com.adx.tencent.oppo.service.OppoRetryService;
-import com.adx.tencent.oppo.service.OppoRetryTask;
 import com.adx.tencent.oppo.service.OppoTagEventSyncService;
 import com.adx.tencent.oppo.service.OppoColdWorker;
 import com.adx.tencent.storage.RedisHotStore;
@@ -186,11 +185,6 @@ public class AppConfiguration {
     }
 
     @Bean
-    public OppoRetryTask oppoRetryTask(@Nullable OppoRetryService retryService) {
-        return retryService == null ? null : new OppoRetryTask(retryService, props);
-    }
-
-    @Bean
     public OppoColdStore oppoColdStore(@Nullable DataSource tidbDataSource) {
         if (tidbDataSource == null) return null;
         try {

src/main/java/com/adx/tencent/oppo/service/OppoBackgroundTasks.java

@@ -1,22 +1,118 @@
 package com.adx.tencent.oppo.service;
 
 import com.adx.tencent.config.AppProperties;
+import com.adx.tencent.leader.LeaderElection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
 import org.springframework.lang.Nullable;
 import org.springframework.stereotype.Component;
+
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 @Component
 public class OppoBackgroundTasks implements ApplicationRunner {
+    private static final Logger log = LoggerFactory.getLogger(OppoBackgroundTasks.class);
+
     @Autowired private AppProperties props;
-    @Autowired(required=false) private OppoColdWorker worker;
-    @Autowired(required=false) private OppoTagEventSyncService tagSync;
-    private final AtomicBoolean stopped=new AtomicBoolean();
-    private final ExecutorService executor=Executors.newCachedThreadPool(r->{Thread t=new Thread(r);t.setDaemon(true);return t;});
-    public void run(ApplicationArguments args){if(worker!=null)executor.submit(()->{while(!stopped.get()){try{worker.runOnce();}catch(Exception e){try{Thread.sleep(1000);}catch(InterruptedException x){Thread.currentThread().interrupt();return;}}}});if(tagSync!=null&&props.isOppoTagEventSyncEnabled())executor.submit(()->{while(!stopped.get()){try{tagSync.syncOnce();}catch(Exception ignored){}sleep(props.getOppoTagEventSyncInterval().toMillis());}});}
-    private void sleep(long millis){try{Thread.sleep(Math.max(1000,millis));}catch(InterruptedException e){Thread.currentThread().interrupt();stopped.set(true);}}
+    @Autowired(required = false) private OppoColdWorker worker;
+    @Autowired(required = false) private OppoRetryService retryService;
+    @Autowired(required = false) private OppoTagEventSyncService tagSync;
+    @Autowired(required = false) @Nullable private LeaderElection leaderElection;
+
+    private final AtomicBoolean stopped = new AtomicBoolean();
+    private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
+        Thread t = new Thread(r);
+        t.setDaemon(true);
+        return t;
+    });
+
+    public void run(ApplicationArguments args) {
+        if (worker != null) {
+            executor.submit(() -> {
+                while (!stopped.get()) {
+                    try {
+                        worker.runOnce();
+                    } catch (Exception e) {
+                        log.error("oppo cold worker: {}", e.getMessage(), e);
+                        sleep(1000);
+                    }
+                }
+            });
+        }
+
+        if (retryService != null && props.isOppoCallbackRetryEnabled()) {
+            if (props.isSkipLeaderElection() || leaderElection == null) {
+                executor.submit(() -> runRetry(() -> stopped.get()));
+            } else {
+                executor.submit(() -> leaderElection.run(
+                        "adx:lock:oppo:callback-retry",
+                        props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
+                        stopped::get,
+                        this::runRetry,
+                        e -> log.error("oppo callback retry leader: {}", e.getMessage(), e)
+                ));
+            }
+        }
+
+        if (tagSync != null && props.isOppoTagEventSyncEnabled()) {
+            if (props.isSkipLeaderElection() || leaderElection == null) {
+                executor.submit(() -> runTagEventSync(() -> stopped.get()));
+            } else {
+                executor.submit(() -> leaderElection.run(
+                        "adx:lock:oppo:tag-event-sync",
+                        props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
+                        stopped::get,
+                        this::runTagEventSync,
+                        e -> log.error("oppo tag event sync leader: {}", e.getMessage(), e)
+                ));
+            }
+        }
+    }
+
+    private void runRetry(LeaderElection.StopSignal stopSignal) {
+        long intervalMs = props.getOppoCallbackRetryInterval().toMillis();
+        while (!stopSignal.isStopped()) {
+            try {
+                retryService.retry(props.getOppoCallbackRetryLimit());
+            } catch (Exception e) {
+                log.error("oppo callback retry: {}", e.getMessage(), e);
+            }
+            sleepResponsive(intervalMs, stopSignal);
+        }
+    }
+
+    private void runTagEventSync(LeaderElection.StopSignal stopSignal) {
+        long intervalMs = props.getOppoTagEventSyncInterval().toMillis();
+        while (!stopSignal.isStopped()) {
+            try {
+                tagSync.syncOnce();
+            } catch (Exception e) {
+                log.error("oppo tag event sync: {}", e.getMessage(), e);
+            }
+            sleepResponsive(intervalMs, stopSignal);
+        }
+    }
+
+    private void sleepResponsive(long millis, LeaderElection.StopSignal stopSignal) {
+        long remaining = millis;
+        while (remaining > 0 && !stopSignal.isStopped()) {
+            long chunk = Math.min(remaining, 1000);
+            sleep(chunk);
+            remaining -= chunk;
+        }
+    }
+
+    private void sleep(long millis) {
+        try {
+            Thread.sleep(Math.max(1, millis));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            stopped.set(true);
+        }
+    }
 }

src/main/java/com/adx/tencent/oppo/service/OppoRetryTask.java

@@ -1,10 +0,0 @@
-package com.adx.tencent.oppo.service;
-import com.adx.tencent.config.AppProperties;
-import org.springframework.scheduling.annotation.Scheduled;
-
-public class OppoRetryTask {
-    private final OppoRetryService retry; private final AppProperties props;
-    public OppoRetryTask(OppoRetryService retry,AppProperties props){this.retry=retry;this.props=props;}
-    @Scheduled(fixedDelayString="${adx.oppo-callback-retry-interval:300s}")
-    public void run(){if(props.isOppoCallbackRetryEnabled())retry.retry(props.getOppoCallbackRetryLimit());}
-}