AppConfiguration.java 44 KB

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