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 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 { if (worker.runOnce() == 0) { sleep(200); } } 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); } } }