AppConfiguration.java 51 KB

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