AppConfiguration.java 48 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997
  1. package com.adx.tencent;
  2. import com.adx.tencent.baidu.AdxClient;
  3. import com.adx.tencent.baidu.AuctionPriceEncoder;
  4. import com.adx.tencent.baidu.ConversionClient;
  5. import com.adx.tencent.config.AppProperties;
  6. import com.adx.tencent.conversionsync.ConversionSyncRunner;
  7. import com.adx.tencent.conversionsync.ConversionSyncService;
  8. import com.adx.tencent.conversionsync.ConversionBackfillJobService;
  9. import com.adx.tencent.conversionsync.ManualTencentCallbackService;
  10. import com.adx.tencent.conversionsync.RetryService;
  11. import com.adx.tencent.httpapi.MediaPlacement;
  12. import com.adx.tencent.honor.client.HonorClient;
  13. import com.adx.tencent.honor.service.HonorColdWorker;
  14. import com.adx.tencent.honor.service.HonorPlacement;
  15. import com.adx.tencent.honor.service.HonorRetryService;
  16. import com.adx.tencent.honor.service.HonorTagEventResolver;
  17. import com.adx.tencent.honor.service.HonorTagEventSyncService;
  18. import com.adx.tencent.honor.store.HonorColdStore;
  19. import com.adx.tencent.honor.store.HonorHotStore;
  20. import com.adx.tencent.kuaishou.client.KuaishouClient;
  21. import com.adx.tencent.kuaishou.service.KuaishouColdWorker;
  22. import com.adx.tencent.kuaishou.service.KuaishouPlacement;
  23. import com.adx.tencent.kuaishou.service.KuaishouRetryService;
  24. import com.adx.tencent.kuaishou.service.KuaishouTagEventResolver;
  25. import com.adx.tencent.kuaishou.service.KuaishouTagEventSyncService;
  26. import com.adx.tencent.kuaishou.store.KuaishouColdStore;
  27. import com.adx.tencent.kuaishou.store.KuaishouHotStore;
  28. import com.adx.tencent.leader.LeaderElection;
  29. import com.adx.tencent.report.AdBidReportStore;
  30. import com.adx.tencent.tagsync.AccountTagEventResolver;
  31. import com.adx.tencent.tagsync.AccountTagEventSyncService;
  32. import com.adx.tencent.tagsync.TagEventResolver;
  33. import com.adx.tencent.tagsync.TagEventSyncService;
  34. import com.adx.tencent.tencent.TencentClient;
  35. import com.adx.tencent.vivo.client.VivoClient;
  36. import com.adx.tencent.vivo.service.VivoAuthService;
  37. import com.adx.tencent.vivo.service.VivoAdvertiserSyncService;
  38. import com.adx.tencent.vivo.service.VivoColdWorker;
  39. import com.adx.tencent.vivo.service.VivoPlacement;
  40. import com.adx.tencent.vivo.service.VivoRetryService;
  41. import com.adx.tencent.vivo.service.VivoTagEventResolver;
  42. import com.adx.tencent.vivo.service.VivoTagEventSyncService;
  43. import com.adx.tencent.vivo.store.VivoColdStore;
  44. import com.adx.tencent.vivo.store.VivoHotStore;
  45. import com.adx.tencent.storage.RedisHotStore;
  46. import com.adx.tencent.storage.RedisLock;
  47. import com.adx.tencent.storage.TiDBColdStore;
  48. import com.adx.tencent.worker.ColdWorker;
  49. import com.fasterxml.jackson.databind.ObjectMapper;
  50. import com.zaxxer.hikari.HikariConfig;
  51. import com.zaxxer.hikari.HikariDataSource;
  52. import io.lettuce.core.ClientOptions;
  53. import io.lettuce.core.SocketOptions;
  54. import io.lettuce.core.protocol.ProtocolVersion;
  55. import org.apache.ibatis.session.SqlSessionFactory;
  56. import org.mybatis.spring.SqlSessionFactoryBean;
  57. import org.slf4j.Logger;
  58. import org.slf4j.LoggerFactory;
  59. import org.springframework.beans.factory.annotation.Autowired;
  60. import org.springframework.boot.ApplicationArguments;
  61. import org.springframework.boot.ApplicationRunner;
  62. import org.springframework.context.annotation.Bean;
  63. import org.springframework.context.annotation.Configuration;
  64. import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
  65. import org.springframework.data.redis.connection.RedisConnectionFactory;
  66. import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
  67. import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
  68. import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
  69. import org.springframework.data.redis.core.StringRedisTemplate;
  70. import org.springframework.lang.Nullable;
  71. import org.springframework.scheduling.annotation.Scheduled;
  72. import org.springframework.stereotype.Component;
  73. import javax.sql.DataSource;
  74. import java.sql.Connection;
  75. import java.time.Duration;
  76. import java.util.HashMap;
  77. import java.util.List;
  78. import java.util.Map;
  79. import java.util.concurrent.ExecutorService;
  80. import java.util.concurrent.Executors;
  81. import java.util.concurrent.atomic.AtomicBoolean;
  82. /**
  83. * 腾讯-百度回传项目 Bean 组装配置类。
  84. * 负责所有核心 Bean 的创建和后台任务(Worker、Leader 选举、定时同步等)。
  85. */
  86. @Configuration
  87. public class AppConfiguration {
  88. private static final Logger log = LoggerFactory.getLogger(AppConfiguration.class);
  89. @Autowired private AppProperties props;
  90. // ─── Redis ───────────────────────────────────────────────────────────────
  91. @Bean
  92. public RedisConnectionFactory redisConnectionFactory() {
  93. String addr = props.getRedisAddr();
  94. String[] parts = addr.split(":", 2);
  95. String host = parts[0];
  96. int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;
  97. RedisStandaloneConfiguration cfg = new RedisStandaloneConfiguration(host, port);
  98. cfg.setDatabase(props.getRedisDb());
  99. if (props.getRedisUsername() != null && !props.getRedisUsername().isBlank()) {
  100. cfg.setUsername(props.getRedisUsername());
  101. }
  102. if (props.getRedisPassword() != null && !props.getRedisPassword().isBlank()) {
  103. cfg.setPassword(props.getRedisPassword());
  104. }
  105. LettuceClientConfiguration clientCfg = LettuceClientConfiguration.builder()
  106. .clientOptions(ClientOptions.builder()
  107. .protocolVersion(ProtocolVersion.RESP2)
  108. .socketOptions(SocketOptions.builder()
  109. .connectTimeout(props.getRedisConnectTimeout())
  110. .build())
  111. .build())
  112. .commandTimeout(props.getRedisCommandTimeout())
  113. .shutdownTimeout(props.getRedisShutdownTimeout())
  114. .build();
  115. LettuceConnectionFactory factory = new LettuceConnectionFactory(cfg, clientCfg);
  116. factory.setValidateConnection(props.isRedisValidateConnection());
  117. factory.setShareNativeConnection(false);
  118. return factory;
  119. }
  120. @Bean
  121. public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
  122. return new StringRedisTemplate(factory);
  123. }
  124. @Bean
  125. public RedisHotStore redisHotStore(StringRedisTemplate redis,
  126. ObjectMapper objectMapper,
  127. @Nullable TiDBColdStore coldStore) {
  128. return new RedisHotStore(redis, objectMapper,
  129. coldStore,
  130. "adx:tencent:",
  131. props.getRedisStream(),
  132. props.getBidTtl(),
  133. props.getRedisStreamMaxLen());
  134. }
  135. @Bean
  136. public RedisLock redisLock(StringRedisTemplate redis) {
  137. return new RedisLock(redis);
  138. }
  139. @Bean
  140. public LeaderElection leaderElection(RedisLock redisLock) {
  141. return new LeaderElection(redisLock);
  142. }
  143. // ─── TiDB(可选)────────────────────────────────────────────────────────
  144. @Bean
  145. public DataSource tidbDataSource() {
  146. String url = props.getTidbUrl();
  147. if (url == null || url.isBlank()) {
  148. log.info("TiDB URL is empty; cold storage disabled");
  149. return null;
  150. }
  151. try {
  152. return createDataSource("adx-tidb", url, props.getTidbUsername(), props.getTidbPassword());
  153. } catch (Exception e) {
  154. log.warn("Failed to open TiDB datasource: {}", e.getMessage(), e);
  155. return null;
  156. }
  157. }
  158. @Bean
  159. public TiDBColdStore tiDBColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
  160. if (tidbDataSource == null) {
  161. log.info("TiDB datasource is empty; cold storage disabled");
  162. return null;
  163. }
  164. try {
  165. SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
  166. TiDBColdStore store = new TiDBColdStore(sqlSessionFactory, objectMapper);
  167. store.migrate();
  168. log.info("TiDB cold store initialized and migrated");
  169. return store;
  170. } catch (Exception e) {
  171. log.warn("Failed to open TiDB: {}; cold storage disabled", e.getMessage(), e);
  172. return null;
  173. }
  174. }
  175. private SqlSessionFactory createSqlSessionFactory(DataSource dataSource) throws Exception {
  176. SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
  177. factoryBean.setDataSource(dataSource);
  178. factoryBean.setMapperLocations(
  179. new PathMatchingResourcePatternResolver().getResources("classpath:mapper/*.xml"));
  180. return factoryBean.getObject();
  181. }
  182. private DataSource createDataSource(String poolName, String jdbcUrl, String username, String password) {
  183. HikariConfig config = new HikariConfig();
  184. config.setDriverClassName("com.mysql.cj.jdbc.Driver");
  185. config.setJdbcUrl(jdbcUrl);
  186. if (username != null && !username.isBlank()) config.setUsername(username);
  187. if (password != null && !password.isBlank()) config.setPassword(password);
  188. config.setPoolName(poolName);
  189. config.setMinimumIdle(props.getTidbMinimumIdle());
  190. config.setMaximumPoolSize(props.getTidbMaximumPoolSize());
  191. config.setConnectionTimeout(props.getTidbConnectionTimeout().toMillis());
  192. config.setValidationTimeout(props.getTidbValidationTimeout().toMillis());
  193. config.setIdleTimeout(props.getTidbIdleTimeout().toMillis());
  194. config.setMaxLifetime(props.getTidbMaxLifetime().toMillis());
  195. config.setKeepaliveTime(props.getTidbKeepaliveTime().toMillis());
  196. return new HikariDataSource(config);
  197. }
  198. // ─── Baidu 客户端 ────────────────────────────────────────────────────────
  199. @Bean
  200. public AuctionPriceEncoder auctionPriceEncoder() {
  201. return AuctionPriceEncoder.create(
  202. props.getBaiduAuctionPriceEncryption(),
  203. props.getBaiduAuctionPriceEKey(),
  204. props.getBaiduAuctionPriceIKey());
  205. }
  206. @Bean
  207. public AdxClient adxClient(AuctionPriceEncoder encoder, ObjectMapper objectMapper) {
  208. return new AdxClient(
  209. props.getBaiduAdxEndpoint(),
  210. encoder,
  211. objectMapper,
  212. props.getBaiduTrackingMaxRedirects(),
  213. props.getBaiduTrackingMax500Retries());
  214. }
  215. @Bean
  216. public ConversionClient conversionClient(ObjectMapper objectMapper) {
  217. return new ConversionClient(
  218. props.getBaiduConversionBaseUrl(),
  219. props.getBaiduCustomerName(),
  220. props.getBaiduSecret(),
  221. objectMapper);
  222. }
  223. @Bean
  224. public HonorColdStore honorColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
  225. if (tidbDataSource == null) return null;
  226. try {
  227. SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
  228. HonorColdStore store = new HonorColdStore(sqlSessionFactory, objectMapper);
  229. store.migrate();
  230. return store;
  231. } catch (Exception e) {
  232. log.warn("Failed to open Honor TiDB: {}", e.getMessage(), e);
  233. return null;
  234. }
  235. }
  236. @Bean
  237. public HonorHotStore honorHotStore(StringRedisTemplate redis,
  238. ObjectMapper objectMapper,
  239. @Nullable HonorColdStore honorColdStore) {
  240. if (honorColdStore == null) return null;
  241. return new HonorHotStore(redis, objectMapper, honorColdStore,
  242. "adx:honor:", props.getHonorRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
  243. }
  244. @Bean
  245. public HonorPlacement honorPlacement() {
  246. HonorPlacement p = new HonorPlacement();
  247. p.setBaiduMediaId(props.getBaiduMediaId());
  248. p.setBaiduAppId(props.getBaiduAppId());
  249. p.setBaiduTagId(props.getBaiduTagId());
  250. p.setBidFloor(props.getBaiduBidFloor());
  251. p.setActionTypes(List.of(0, 1, 2));
  252. Map<String, HonorPlacement.HonorAppPlacement> platforms = new HashMap<>();
  253. String honorAndroidAppId = props.getBaiduAndroidAppId();
  254. List<String> honorAndroidTagIds = props.getBaiduAndroidTagIds();
  255. if (honorAndroidAppId != null && !honorAndroidAppId.isBlank()) {
  256. HonorPlacement.HonorAppPlacement android = new HonorPlacement.HonorAppPlacement();
  257. android.setAppId(honorAndroidAppId);
  258. android.setTagIds(honorAndroidTagIds);
  259. platforms.put("android", android);
  260. }
  261. String honorIosAppId = props.getBaiduIosAppId();
  262. List<String> honorIosTagIds = props.getBaiduIosTagIds();
  263. if (honorIosAppId != null && !honorIosAppId.isBlank()) {
  264. HonorPlacement.HonorAppPlacement ios = new HonorPlacement.HonorAppPlacement();
  265. ios.setAppId(honorIosAppId);
  266. ios.setTagIds(honorIosTagIds);
  267. platforms.put("ios", ios);
  268. }
  269. if (!platforms.isEmpty()) p.setPlatforms(platforms);
  270. return p;
  271. }
  272. @Bean
  273. public HonorClient honorClient(ObjectMapper objectMapper) {
  274. return new HonorClient(props.getHonorConversionBaseUrl(), props.getHonorChannelId(), objectMapper);
  275. }
  276. @Bean
  277. public HonorRetryService honorRetryService(@Nullable HonorColdStore honorColdStore, HonorClient honorClient) {
  278. if (honorColdStore == null) return null;
  279. return new HonorRetryService(honorColdStore, honorClient, props.getHonorCallbackRetryLimit());
  280. }
  281. @Bean
  282. public HonorColdWorker honorColdWorker(@Nullable HonorHotStore honorHotStore,
  283. @Nullable HonorColdStore honorColdStore,
  284. ObjectMapper objectMapper) {
  285. if (honorHotStore == null || honorColdStore == null) return null;
  286. return new HonorColdWorker(honorHotStore, honorColdStore,
  287. "honor-cold-writers", props.getWorkerConsumer() + "-honor", props.getWorkerBatch(),
  288. null, props.getRedisStreamMaxLen(), objectMapper);
  289. }
  290. @Bean
  291. public KuaishouColdStore kuaishouColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
  292. if (tidbDataSource == null) return null;
  293. try {
  294. SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
  295. KuaishouColdStore store = new KuaishouColdStore(sqlSessionFactory, objectMapper);
  296. store.migrate();
  297. return store;
  298. } catch (Exception e) {
  299. log.warn("Failed to open Kuaishou TiDB: {}", e.getMessage(), e);
  300. return null;
  301. }
  302. }
  303. @Bean
  304. public KuaishouHotStore kuaishouHotStore(StringRedisTemplate redis,
  305. ObjectMapper objectMapper,
  306. @Nullable KuaishouColdStore kuaishouColdStore) {
  307. if (kuaishouColdStore == null) return null;
  308. return new KuaishouHotStore(redis, objectMapper, kuaishouColdStore,
  309. "adx:kuaishou:", props.getKuaishouRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
  310. }
  311. @Bean
  312. public KuaishouPlacement kuaishouPlacement() {
  313. KuaishouPlacement p = new KuaishouPlacement();
  314. p.setBaiduMediaId(props.getBaiduMediaId());
  315. p.setBaiduAppId(props.getBaiduAppId());
  316. p.setBaiduTagId(props.getBaiduTagId());
  317. p.setBidFloor(props.getBaiduBidFloor());
  318. p.setActionTypes(List.of(0, 1, 2));
  319. Map<String, KuaishouPlacement.KuaishouAppPlacement> platforms = new HashMap<>();
  320. if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
  321. KuaishouPlacement.KuaishouAppPlacement android = new KuaishouPlacement.KuaishouAppPlacement();
  322. android.setAppId(props.getBaiduAndroidAppId());
  323. android.setTagIds(props.getBaiduAndroidTagIds());
  324. platforms.put("android", android);
  325. }
  326. if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
  327. KuaishouPlacement.KuaishouAppPlacement ios = new KuaishouPlacement.KuaishouAppPlacement();
  328. ios.setAppId(props.getBaiduIosAppId());
  329. ios.setTagIds(props.getBaiduIosTagIds());
  330. platforms.put("ios", ios);
  331. }
  332. if (!platforms.isEmpty()) p.setPlatforms(platforms);
  333. return p;
  334. }
  335. @Bean
  336. public KuaishouClient kuaishouClient(ObjectMapper objectMapper) {
  337. return new KuaishouClient(props.getKuaishouConversionBaseUrl(), objectMapper);
  338. }
  339. @Bean
  340. public KuaishouRetryService kuaishouRetryService(@Nullable KuaishouColdStore kuaishouColdStore,
  341. KuaishouClient kuaishouClient) {
  342. if (kuaishouColdStore == null) return null;
  343. return new KuaishouRetryService(kuaishouColdStore, kuaishouClient, props.getKuaishouCallbackRetryLimit());
  344. }
  345. @Bean
  346. public KuaishouColdWorker kuaishouColdWorker(@Nullable KuaishouHotStore kuaishouHotStore,
  347. @Nullable KuaishouColdStore kuaishouColdStore,
  348. ObjectMapper objectMapper) {
  349. if (kuaishouHotStore == null || kuaishouColdStore == null) return null;
  350. return new KuaishouColdWorker(kuaishouHotStore, kuaishouColdStore,
  351. "kuaishou-cold-writers", props.getWorkerConsumer() + "-kuaishou", props.getWorkerBatch(),
  352. null, props.getRedisStreamMaxLen(), objectMapper);
  353. }
  354. @Bean
  355. public VivoColdStore vivoColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
  356. if (tidbDataSource == null) return null;
  357. try {
  358. SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
  359. VivoColdStore store = new VivoColdStore(sqlSessionFactory, objectMapper);
  360. store.migrate();
  361. return store;
  362. } catch (Exception e) {
  363. log.warn("Failed to open Vivo TiDB: {}", e.getMessage(), e);
  364. return null;
  365. }
  366. }
  367. @Bean
  368. public VivoHotStore vivoHotStore(StringRedisTemplate redis,
  369. ObjectMapper objectMapper,
  370. @Nullable VivoColdStore vivoColdStore) {
  371. if (vivoColdStore == null) return null;
  372. return new VivoHotStore(redis, objectMapper, vivoColdStore,
  373. "adx:vivo:", props.getVivoRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
  374. }
  375. @Bean
  376. public VivoPlacement vivoPlacement() {
  377. VivoPlacement p = new VivoPlacement();
  378. p.setBaiduMediaId(props.getBaiduMediaId());
  379. p.setBaiduAppId(props.getBaiduAppId());
  380. p.setBaiduTagId(props.getBaiduTagId());
  381. p.setBidFloor(props.getBaiduBidFloor());
  382. p.setActionTypes(List.of(0, 1, 2));
  383. Map<String, VivoPlacement.VivoAppPlacement> platforms = new HashMap<>();
  384. if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
  385. VivoPlacement.VivoAppPlacement android = new VivoPlacement.VivoAppPlacement();
  386. android.setAppId(props.getBaiduAndroidAppId());
  387. android.setTagIds(props.getBaiduAndroidTagIds());
  388. platforms.put("android", android);
  389. }
  390. if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
  391. VivoPlacement.VivoAppPlacement ios = new VivoPlacement.VivoAppPlacement();
  392. ios.setAppId(props.getBaiduIosAppId());
  393. ios.setTagIds(props.getBaiduIosTagIds());
  394. platforms.put("ios", ios);
  395. }
  396. if (!platforms.isEmpty()) p.setPlatforms(platforms);
  397. return p;
  398. }
  399. @Bean
  400. public VivoAuthService vivoAuthService(StringRedisTemplate redisTemplate,
  401. @Nullable VivoColdStore vivoColdStore,
  402. ObjectMapper objectMapper) {
  403. if (vivoColdStore == null) return null;
  404. return new VivoAuthService(
  405. props.getVivoConversionBaseUrl(),
  406. props.getVivoRedirectUri(),
  407. props.getVivoTokenRedisPrefix(),
  408. props.getVivoAdvertiserRedisPrefix(),
  409. redisTemplate,
  410. vivoColdStore,
  411. objectMapper
  412. );
  413. }
  414. @Bean
  415. public VivoClient vivoClient(@Nullable VivoAuthService vivoAuthService, ObjectMapper objectMapper) {
  416. if (vivoAuthService == null) return null;
  417. return new VivoClient(props.getVivoConversionBaseUrl(), vivoAuthService, objectMapper);
  418. }
  419. @Bean
  420. public VivoRetryService vivoRetryService(@Nullable VivoColdStore vivoColdStore,
  421. @Nullable VivoClient vivoClient) {
  422. if (vivoColdStore == null || vivoClient == null) return null;
  423. return new VivoRetryService(vivoColdStore, vivoClient, props.getVivoCallbackRetryLimit());
  424. }
  425. @Bean
  426. public VivoAdvertiserSyncService vivoAdvertiserSyncService(@Nullable VivoAuthService vivoAuthService,
  427. @Nullable VivoColdStore vivoColdStore,
  428. ObjectMapper objectMapper) {
  429. if (vivoAuthService == null || vivoColdStore == null) return null;
  430. return new VivoAdvertiserSyncService(props.getVivoConversionBaseUrl(), vivoAuthService, vivoColdStore, objectMapper);
  431. }
  432. @Bean
  433. public VivoColdWorker vivoColdWorker(@Nullable VivoHotStore vivoHotStore,
  434. @Nullable VivoColdStore vivoColdStore,
  435. ObjectMapper objectMapper) {
  436. if (vivoHotStore == null || vivoColdStore == null) return null;
  437. return new VivoColdWorker(vivoHotStore, vivoColdStore,
  438. "vivo-cold-writers", props.getWorkerConsumer() + "-vivo", props.getWorkerBatch(),
  439. null, props.getRedisStreamMaxLen(), objectMapper);
  440. }
  441. // ─── Tencent 客户端 + Sync 服务 ──────────────────────────────────────────
  442. @Bean
  443. public TencentClient tencentClient(ObjectMapper objectMapper, StringRedisTemplate redisTemplate) {
  444. // token 从 Redis 动态获取,支持外部刷新
  445. String tokenRedisKey = props.getTencentAccessTokenRedisKey();
  446. java.util.function.Supplier<String> tokenSupplier;
  447. if (tokenRedisKey != null && !tokenRedisKey.isBlank()) {
  448. tokenSupplier = () -> redisTemplate.opsForValue().get(tokenRedisKey);
  449. } else {
  450. // 兜底:使用配置文件中的静态 token
  451. String staticToken = props.getTencentAccessToken();
  452. tokenSupplier = () -> staticToken;
  453. }
  454. return new TencentClient(
  455. tokenSupplier,
  456. props.getTencentActMap(),
  457. objectMapper);
  458. }
  459. @Bean
  460. public TagEventResolver tagEventResolver(@Nullable TiDBColdStore coldStore,
  461. StringRedisTemplate redisTemplate) {
  462. if (coldStore == null) return null;
  463. return new TagEventResolver(redisTemplate, coldStore, props.getTagEventSyncRedisPrefix());
  464. }
  465. @Bean
  466. public AccountTagEventResolver accountTagEventResolver(@Nullable TiDBColdStore coldStore,
  467. StringRedisTemplate redisTemplate) {
  468. if (coldStore == null) return null;
  469. return new AccountTagEventResolver(redisTemplate, coldStore, props.getAccountTagEventSyncRedisPrefix());
  470. }
  471. @Bean
  472. public HonorTagEventResolver honorTagEventResolver(@Nullable HonorColdStore honorColdStore,
  473. StringRedisTemplate redisTemplate) {
  474. if (honorColdStore == null) return null;
  475. return new HonorTagEventResolver(redisTemplate, honorColdStore, props.getHonorTagEventSyncRedisPrefix());
  476. }
  477. @Bean
  478. public KuaishouTagEventResolver kuaishouTagEventResolver(@Nullable KuaishouColdStore kuaishouColdStore,
  479. StringRedisTemplate redisTemplate) {
  480. if (kuaishouColdStore == null) return null;
  481. return new KuaishouTagEventResolver(redisTemplate, kuaishouColdStore, props.getKuaishouTagEventSyncRedisPrefix());
  482. }
  483. @Bean
  484. public VivoTagEventResolver vivoTagEventResolver(@Nullable VivoColdStore vivoColdStore,
  485. StringRedisTemplate redisTemplate) {
  486. if (vivoColdStore == null) return null;
  487. return new VivoTagEventResolver(redisTemplate, vivoColdStore, props.getVivoTagEventSyncRedisPrefix());
  488. }
  489. @Bean
  490. public ConversionSyncService conversionSyncService(ConversionClient baiduClient,
  491. RedisHotStore hotStore,
  492. TencentClient tencentClient,
  493. @Nullable TiDBColdStore coldStore,
  494. @Nullable TagEventResolver tagEventResolver,
  495. @Nullable AccountTagEventResolver accountTagEventResolver,
  496. @Nullable HonorTagEventResolver honorTagEventResolver,
  497. @Nullable KuaishouTagEventResolver kuaishouTagEventResolver,
  498. @Nullable VivoTagEventResolver vivoTagEventResolver,
  499. @Nullable HonorHotStore honorHotStore,
  500. @Nullable HonorColdStore honorColdStore,
  501. @Nullable KuaishouHotStore kuaishouHotStore,
  502. @Nullable KuaishouColdStore kuaishouColdStore,
  503. @Nullable VivoHotStore vivoHotStore,
  504. @Nullable VivoColdStore vivoColdStore,
  505. HonorClient honorClient,
  506. KuaishouClient kuaishouClient,
  507. @Nullable VivoClient vivoClient) {
  508. if (coldStore == null) return null;
  509. return new ConversionSyncService(baiduClient, hotStore, tencentClient, coldStore, tagEventResolver,
  510. accountTagEventResolver, honorTagEventResolver, kuaishouTagEventResolver, vivoTagEventResolver,
  511. honorHotStore, honorColdStore, kuaishouHotStore, kuaishouColdStore, vivoHotStore, vivoColdStore,
  512. honorClient, kuaishouClient, vivoClient, props);
  513. }
  514. @Bean
  515. public ConversionSyncRunner conversionSyncRunner(@Nullable ConversionSyncService syncer) {
  516. if (syncer == null) return null;
  517. return new ConversionSyncRunner(syncer,
  518. props.getConversionSyncDateOffsetDays(),
  519. props.getConversionSyncPageSize(),
  520. props.getConversionSyncActs());
  521. }
  522. @Bean
  523. public ConversionBackfillJobService conversionBackfillJobService(@Nullable ConversionSyncRunner syncer) {
  524. if (syncer == null) return null;
  525. return new ConversionBackfillJobService(syncer);
  526. }
  527. @Bean
  528. public RetryService retryService(@Nullable TiDBColdStore coldStore, TencentClient tencentClient) {
  529. if (coldStore == null) return null;
  530. return new RetryService(coldStore, tencentClient, props.getCallbackRetryLimit());
  531. }
  532. @Bean
  533. public ManualTencentCallbackService manualTencentCallbackService(@Nullable TiDBColdStore coldStore,
  534. RedisHotStore hotStore,
  535. TencentClient tencentClient) {
  536. if (coldStore == null) return null;
  537. return new ManualTencentCallbackService(coldStore, hotStore, tencentClient);
  538. }
  539. @Bean
  540. public TagEventSyncService tagEventSyncService(@Nullable TiDBColdStore coldStore,
  541. StringRedisTemplate redisTemplate) {
  542. if (coldStore == null) return null;
  543. // TTL 取同步间隔的 3 倍,避免表中已删除的 tag 在 Redis 长期残留
  544. Duration ttl = props.getTagEventSyncInterval().multipliedBy(3);
  545. return new TagEventSyncService(coldStore, redisTemplate,
  546. props.getTagEventSyncRedisPrefix(), ttl);
  547. }
  548. @Bean
  549. public AccountTagEventSyncService accountTagEventSyncService(@Nullable TiDBColdStore coldStore,
  550. StringRedisTemplate redisTemplate) {
  551. if (coldStore == null) return null;
  552. Duration ttl = props.getAccountTagEventSyncInterval().multipliedBy(3);
  553. return new AccountTagEventSyncService(coldStore, redisTemplate,
  554. props.getAccountTagEventSyncRedisPrefix(), ttl);
  555. }
  556. @Bean
  557. public HonorTagEventSyncService honorTagEventSyncService(@Nullable HonorColdStore honorColdStore,
  558. StringRedisTemplate redisTemplate) {
  559. if (honorColdStore == null) return null;
  560. Duration ttl = props.getHonorTagEventSyncInterval().multipliedBy(3);
  561. return new HonorTagEventSyncService(honorColdStore, redisTemplate,
  562. props.getHonorTagEventSyncRedisPrefix(), ttl);
  563. }
  564. @Bean
  565. public KuaishouTagEventSyncService kuaishouTagEventSyncService(@Nullable KuaishouColdStore kuaishouColdStore,
  566. StringRedisTemplate redisTemplate) {
  567. if (kuaishouColdStore == null) return null;
  568. Duration ttl = props.getKuaishouTagEventSyncInterval().multipliedBy(3);
  569. return new KuaishouTagEventSyncService(kuaishouColdStore, redisTemplate,
  570. props.getKuaishouTagEventSyncRedisPrefix(), ttl);
  571. }
  572. @Bean
  573. public VivoTagEventSyncService vivoTagEventSyncService(@Nullable VivoColdStore vivoColdStore,
  574. StringRedisTemplate redisTemplate) {
  575. if (vivoColdStore == null) return null;
  576. Duration ttl = props.getVivoTagEventSyncInterval().multipliedBy(3);
  577. return new VivoTagEventSyncService(vivoColdStore, redisTemplate,
  578. props.getVivoTagEventSyncRedisPrefix(), ttl);
  579. }
  580. @Bean
  581. public AdBidReportStore adBidReportStore(@Nullable DataSource tidbDataSource) {
  582. if (tidbDataSource == null) return null;
  583. String mainSchema = resolveDatabaseName(tidbDataSource, props.getTidbUrl());
  584. if (mainSchema == null || mainSchema.isBlank()) return null;
  585. AdBidReportStore store = new AdBidReportStore(
  586. tidbDataSource,
  587. mainSchema,
  588. props.getReportDatabaseName(),
  589. props.getAdBidReportRetentionDays(),
  590. props.getTrackingReportRetentionDays(),
  591. props.getAdBidReportArchiveBatchSize(),
  592. props.getAdBidReportArchiveMaxBatchesPerRun(),
  593. props.getAdBidReportStatsLookbackDays());
  594. store.migrate();
  595. return store;
  596. }
  597. // ─── ColdWorker ─────────────────────────────────────────────────────────
  598. @Bean
  599. public ColdWorker coldWorker(RedisHotStore hotStore, @Nullable TiDBColdStore coldStore,
  600. ObjectMapper objectMapper) {
  601. if (coldStore == null) return null;
  602. return new ColdWorker(hotStore, coldStore,
  603. props.getWorkerGroup(), props.getWorkerConsumer(),
  604. props.getWorkerBatch(), null,
  605. props.getRedisStreamMaxLen(), objectMapper);
  606. }
  607. // ─── MediaPlacement(腾讯)───────────────────────────────────────────────
  608. @Bean
  609. public MediaPlacement tencentPlacement() {
  610. MediaPlacement p = new MediaPlacement();
  611. p.setBaiduMediaId(props.getBaiduMediaId());
  612. p.setBaiduAppId(props.getBaiduAppId());
  613. p.setBaiduTagId(props.getBaiduTagId());
  614. p.setBidFloor(props.getBaiduBidFloor());
  615. p.setActionTypes(List.of(0, 1, 2));
  616. Map<String, MediaPlacement.BaiduAppPlacement> platforms = new HashMap<>();
  617. if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
  618. MediaPlacement.BaiduAppPlacement android = new MediaPlacement.BaiduAppPlacement();
  619. android.setAppId(props.getBaiduAndroidAppId());
  620. android.setTagIds(props.getBaiduAndroidTagIds());
  621. platforms.put("android", android);
  622. }
  623. if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
  624. MediaPlacement.BaiduAppPlacement ios = new MediaPlacement.BaiduAppPlacement();
  625. ios.setAppId(props.getBaiduIosAppId());
  626. ios.setTagIds(props.getBaiduIosTagIds());
  627. platforms.put("ios", ios);
  628. }
  629. if (!platforms.isEmpty()) p.setPlatforms(platforms);
  630. return p;
  631. }
  632. private static String resolveDatabaseName(DataSource dataSource, String jdbcUrl) {
  633. try (Connection connection = dataSource.getConnection()) {
  634. String catalog = connection.getCatalog();
  635. if (catalog != null && !catalog.isBlank()) {
  636. return catalog.trim();
  637. }
  638. } catch (Exception ignored) {
  639. }
  640. return extractDatabaseName(jdbcUrl);
  641. }
  642. private static String extractDatabaseName(String jdbcUrl) {
  643. if (jdbcUrl == null || jdbcUrl.isBlank()) {
  644. return null;
  645. }
  646. String value = jdbcUrl.trim();
  647. int scheme = value.indexOf("://");
  648. int searchFrom = scheme >= 0 ? scheme + 3 : 0;
  649. int slash = value.indexOf('/', searchFrom);
  650. if (slash < 0 || slash + 1 >= value.length()) {
  651. return null;
  652. }
  653. int end = value.length();
  654. int question = value.indexOf('?', slash + 1);
  655. if (question >= 0) {
  656. end = Math.min(end, question);
  657. }
  658. int hash = value.indexOf('#', slash + 1);
  659. if (hash >= 0) {
  660. end = Math.min(end, hash);
  661. }
  662. int semicolon = value.indexOf(';', slash + 1);
  663. if (semicolon >= 0) {
  664. end = Math.min(end, semicolon);
  665. }
  666. int nextSlash = value.indexOf('/', slash + 1);
  667. if (nextSlash >= 0 && nextSlash < end) {
  668. end = nextSlash;
  669. }
  670. String db = value.substring(slash + 1, end);
  671. return db == null || db.isBlank() ? null : db.trim();
  672. }
  673. // ─── 后台任务调度 ────────────────────────────────────────────────────────
  674. @Component
  675. static class BackgroundTasks implements ApplicationRunner {
  676. private static final Logger taskLog = LoggerFactory.getLogger(BackgroundTasks.class);
  677. @Autowired private AppProperties props;
  678. @Autowired(required = false) private ColdWorker coldWorker;
  679. @Autowired(required = false) private ConversionSyncRunner conversionSyncRunner;
  680. @Autowired(required = false) private RetryService retryService;
  681. @Autowired(required = false) private TagEventSyncService tagEventSyncService;
  682. @Autowired(required = false) private AccountTagEventSyncService accountTagEventSyncService;
  683. @Autowired(required = false) private LeaderElection leaderElection;
  684. private final AtomicBoolean stopped = new AtomicBoolean(false);
  685. private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
  686. Thread t = new Thread(r);
  687. t.setDaemon(true);
  688. return t;
  689. });
  690. @Override
  691. public void run(ApplicationArguments args) {
  692. // 冷路径 Worker
  693. if (coldWorker != null) {
  694. executor.submit(() -> {
  695. while (!stopped.get()) {
  696. try {
  697. coldWorker.runOnce();
  698. } catch (Exception e) {
  699. taskLog.error("cold worker: {}", e.getMessage(), e);
  700. sleep(1000);
  701. }
  702. }
  703. });
  704. taskLog.info("Cold worker started");
  705. }
  706. // 转化同步
  707. if (conversionSyncRunner != null && props.isConversionSyncEnabled()) {
  708. if (props.isSkipLeaderElection()) {
  709. executor.submit(() -> runConversionSync(() -> stopped.get()));
  710. taskLog.info("Conversion sync started (skip leader election)");
  711. } else if (leaderElection != null) {
  712. executor.submit(() ->
  713. leaderElection.run(
  714. "adx:lock:tencent:conversion-sync",
  715. props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
  716. stopped::get,
  717. (jobStop) -> runConversionSync(jobStop),
  718. e -> taskLog.error("tencent conversion sync leader: {}", e.getMessage(), e)
  719. )
  720. );
  721. taskLog.info("Conversion sync leader election started");
  722. }
  723. } else {
  724. taskLog.warn("Conversion sync NOT started: runner={}, enabled={}",
  725. conversionSyncRunner != null, props.isConversionSyncEnabled());
  726. }
  727. // 回调重试
  728. if (retryService != null && props.isCallbackRetryEnabled()) {
  729. if (props.isSkipLeaderElection()) {
  730. executor.submit(() -> runCallbackRetry(() -> stopped.get()));
  731. taskLog.info("Callback retry started (skip leader election)");
  732. } else if (leaderElection != null) {
  733. executor.submit(() ->
  734. leaderElection.run(
  735. "adx:lock:tencent:callback-retry",
  736. props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
  737. stopped::get,
  738. (jobStop) -> runCallbackRetry(jobStop),
  739. e -> taskLog.error("tencent callback retry leader: {}", e.getMessage(), e)
  740. )
  741. );
  742. taskLog.info("Callback retry leader election started");
  743. }
  744. } else {
  745. taskLog.warn("Callback retry NOT started: retryService={}, enabled={}",
  746. retryService != null, props.isCallbackRetryEnabled());
  747. }
  748. // 广告位回传方式同步
  749. if (tagEventSyncService != null && props.isTagEventSyncEnabled()) {
  750. if (props.isSkipLeaderElection()) {
  751. executor.submit(() -> runTagEventSync(() -> stopped.get()));
  752. taskLog.info("Tag event sync started (skip leader election)");
  753. } else if (leaderElection != null) {
  754. executor.submit(() ->
  755. leaderElection.run(
  756. "adx:lock:tencent:tag-event-sync",
  757. props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
  758. stopped::get,
  759. (jobStop) -> runTagEventSync(jobStop),
  760. e -> taskLog.error("tencent tag event sync leader: {}", e.getMessage(), e)
  761. )
  762. );
  763. taskLog.info("Tag event sync leader election started");
  764. }
  765. } else {
  766. taskLog.warn("Tag event sync NOT started: service={}, enabled={}",
  767. tagEventSyncService != null, props.isTagEventSyncEnabled());
  768. }
  769. // 腾讯账户级广告位回传方式同步
  770. if (accountTagEventSyncService != null && props.isAccountTagEventSyncEnabled()) {
  771. if (props.isSkipLeaderElection()) {
  772. executor.submit(() -> runAccountTagEventSync(() -> stopped.get()));
  773. taskLog.info("Account tag event sync started (skip leader election)");
  774. } else if (leaderElection != null) {
  775. executor.submit(() ->
  776. leaderElection.run(
  777. "adx:lock:tencent:account-tag-event-sync",
  778. props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
  779. stopped::get,
  780. (jobStop) -> runAccountTagEventSync(jobStop),
  781. e -> taskLog.error("tencent account tag event sync leader: {}", e.getMessage(), e)
  782. )
  783. );
  784. taskLog.info("Account tag event sync leader election started");
  785. }
  786. } else {
  787. taskLog.warn("Account tag event sync NOT started: service={}, enabled={}",
  788. accountTagEventSyncService != null, props.isAccountTagEventSyncEnabled());
  789. }
  790. }
  791. private void runConversionSync(LeaderElection.StopSignal jobStop) {
  792. long intervalMs = props.getConversionSyncInterval().toMillis();
  793. taskLog.info("[ConversionSync] task started, interval={}ms", intervalMs);
  794. while (!jobStop.isStopped()) {
  795. try {
  796. taskLog.info("[ConversionSync] executing...");
  797. ConversionSyncService.SyncResult r = conversionSyncRunner.runOnce();
  798. taskLog.info("[ConversionSync] done: fetched={} sent={} failed={}", r.fetched.get(), r.sent.get(), r.failed.get());
  799. } catch (Exception e) {
  800. taskLog.error("[ConversionSync] error: {}", e.getMessage(), e);
  801. }
  802. taskLog.info("[ConversionSync] sleeping {}ms until next run", intervalMs);
  803. sleepResponsive(intervalMs, jobStop);
  804. }
  805. taskLog.info("[ConversionSync] task stopped");
  806. }
  807. @Scheduled(cron = "0 17 1,6,8,10,12,16,20,23 * * ?")
  808. public void runDelayedConversionSync() {
  809. if (conversionSyncRunner == null || !props.isConversionSyncEnabled()) {
  810. return;
  811. }
  812. if (stopped.get()) {
  813. return;
  814. }
  815. Runnable job = () -> {
  816. long startNs = System.nanoTime();
  817. try {
  818. taskLog.info("[ConversionBackfill] executing for offsets=1..7");
  819. ConversionSyncService.SyncResult r = conversionSyncRunner.runBackfillForOffsetsParallel(
  820. List.of(-1, -2, -3, -4, -5, -6, -7), 3);
  821. long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
  822. taskLog.info("[ConversionBackfill] done: fetched={} matched={} sent={} failed={} skipped={} alreadySent={}",
  823. r.fetched.get(), r.matched.get(), r.sent.get(), r.failed.get(),
  824. r.skipped.get(), r.alreadySent.get());
  825. taskLog.info("[ConversionBackfill] total cost={}ms", costMs);
  826. } catch (Exception e) {
  827. long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
  828. taskLog.error("[ConversionBackfill] error after {}ms: {}", costMs, e.getMessage(), e);
  829. }
  830. };
  831. if (props.isSkipLeaderElection() || leaderElection == null) {
  832. job.run();
  833. return;
  834. }
  835. executor.submit(() ->
  836. leaderElection.runOnce(
  837. "adx:lock:tencent:conversion-backfill",
  838. props.getTaskLockTtl(), props.getTaskLockRenewInterval(),
  839. stopped::get,
  840. stop -> job.run(),
  841. e -> taskLog.error("tencent conversion backfill leader: {}", e.getMessage(), e)
  842. )
  843. );
  844. }
  845. private void runCallbackRetry(LeaderElection.StopSignal jobStop) {
  846. long intervalMs = props.getCallbackRetryInterval().toMillis();
  847. taskLog.info("[CallbackRetry] task started, interval={}ms, limit={}", intervalMs, props.getCallbackRetryLimit());
  848. while (!jobStop.isStopped()) {
  849. try {
  850. taskLog.info("[CallbackRetry] executing...");
  851. RetryService.RetryResult r = retryService.retryTencentCallbacks(props.getCallbackRetryLimit());
  852. taskLog.info("[CallbackRetry] done: fetched={} sent={} failed={}", r.fetched, r.sent, r.failed);
  853. } catch (Exception e) {
  854. taskLog.error("[CallbackRetry] error: {}", e.getMessage(), e);
  855. }
  856. taskLog.info("[CallbackRetry] sleeping {}ms until next run", intervalMs);
  857. sleepResponsive(intervalMs, jobStop);
  858. }
  859. taskLog.info("[CallbackRetry] task stopped");
  860. }
  861. private void runTagEventSync(LeaderElection.StopSignal jobStop) {
  862. long intervalMs = props.getTagEventSyncInterval().toMillis();
  863. taskLog.info("[TagEventSync] task started, interval={}ms", intervalMs);
  864. while (!jobStop.isStopped()) {
  865. try {
  866. int n = tagEventSyncService.syncOnce();
  867. taskLog.info("[TagEventSync] done: synced={}", n);
  868. } catch (Exception e) {
  869. taskLog.error("[TagEventSync] error: {}", e.getMessage(), e);
  870. }
  871. sleepResponsive(intervalMs, jobStop);
  872. }
  873. taskLog.info("[TagEventSync] task stopped");
  874. }
  875. private void runAccountTagEventSync(LeaderElection.StopSignal jobStop) {
  876. long intervalMs = props.getAccountTagEventSyncInterval().toMillis();
  877. taskLog.info("[AccountTagEventSync] task started, interval={}ms", intervalMs);
  878. while (!jobStop.isStopped()) {
  879. try {
  880. int n = accountTagEventSyncService.syncOnce();
  881. taskLog.info("[AccountTagEventSync] done: synced={}", n);
  882. } catch (Exception e) {
  883. taskLog.error("[AccountTagEventSync] error: {}", e.getMessage(), e);
  884. }
  885. sleepResponsive(intervalMs, jobStop);
  886. }
  887. taskLog.info("[AccountTagEventSync] task stopped");
  888. }
  889. private static void sleepResponsive(long ms, LeaderElection.StopSignal jobStop) {
  890. long deadline = System.currentTimeMillis() + ms;
  891. while (!jobStop.isStopped()) {
  892. long remaining = deadline - System.currentTimeMillis();
  893. if (remaining <= 0) break;
  894. try {
  895. Thread.sleep(Math.min(remaining, 200));
  896. } catch (InterruptedException e) {
  897. Thread.currentThread().interrupt();
  898. break;
  899. }
  900. }
  901. }
  902. private static void sleep(long ms) {
  903. try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
  904. }
  905. }
  906. }