| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933 |
- package com.adx.tencent;
- import com.adx.tencent.baidu.AdxClient;
- import com.adx.tencent.baidu.AuctionPriceEncoder;
- import com.adx.tencent.baidu.ConversionClient;
- import com.adx.tencent.config.AppProperties;
- import com.adx.tencent.conversionsync.ConversionSyncRunner;
- import com.adx.tencent.conversionsync.ConversionSyncService;
- import com.adx.tencent.conversionsync.ConversionBackfillJobService;
- import com.adx.tencent.conversionsync.ManualTencentCallbackService;
- import com.adx.tencent.conversionsync.RetryService;
- import com.adx.tencent.httpapi.MediaPlacement;
- import com.adx.tencent.honor.client.HonorClient;
- import com.adx.tencent.honor.service.HonorColdWorker;
- import com.adx.tencent.honor.service.HonorPlacement;
- import com.adx.tencent.honor.service.HonorRetryService;
- import com.adx.tencent.honor.service.HonorTagEventResolver;
- import com.adx.tencent.honor.service.HonorTagEventSyncService;
- import com.adx.tencent.honor.store.HonorColdStore;
- import com.adx.tencent.honor.store.HonorHotStore;
- import com.adx.tencent.kuaishou.client.KuaishouClient;
- import com.adx.tencent.kuaishou.service.KuaishouColdWorker;
- import com.adx.tencent.kuaishou.service.KuaishouPlacement;
- import com.adx.tencent.kuaishou.service.KuaishouRetryService;
- import com.adx.tencent.kuaishou.service.KuaishouTagEventResolver;
- import com.adx.tencent.kuaishou.service.KuaishouTagEventSyncService;
- import com.adx.tencent.kuaishou.store.KuaishouColdStore;
- import com.adx.tencent.kuaishou.store.KuaishouHotStore;
- import com.adx.tencent.leader.LeaderElection;
- import com.adx.tencent.report.AdBidReportStore;
- import com.adx.tencent.tagsync.TagEventResolver;
- import com.adx.tencent.tagsync.TagEventSyncService;
- import com.adx.tencent.tencent.TencentClient;
- import com.adx.tencent.vivo.client.VivoClient;
- import com.adx.tencent.vivo.service.VivoAuthService;
- import com.adx.tencent.vivo.service.VivoAdvertiserSyncService;
- import com.adx.tencent.vivo.service.VivoColdWorker;
- import com.adx.tencent.vivo.service.VivoPlacement;
- import com.adx.tencent.vivo.service.VivoRetryService;
- import com.adx.tencent.vivo.service.VivoTagEventResolver;
- import com.adx.tencent.vivo.service.VivoTagEventSyncService;
- import com.adx.tencent.vivo.store.VivoColdStore;
- import com.adx.tencent.vivo.store.VivoHotStore;
- import com.adx.tencent.storage.RedisHotStore;
- import com.adx.tencent.storage.RedisLock;
- import com.adx.tencent.storage.TiDBColdStore;
- import com.adx.tencent.worker.ColdWorker;
- import com.fasterxml.jackson.databind.ObjectMapper;
- import com.zaxxer.hikari.HikariConfig;
- import com.zaxxer.hikari.HikariDataSource;
- import io.lettuce.core.ClientOptions;
- import io.lettuce.core.SocketOptions;
- import io.lettuce.core.protocol.ProtocolVersion;
- import org.apache.ibatis.session.SqlSessionFactory;
- import org.mybatis.spring.SqlSessionFactoryBean;
- import org.slf4j.Logger;
- import org.slf4j.LoggerFactory;
- import org.springframework.beans.factory.annotation.Autowired;
- import org.springframework.boot.ApplicationArguments;
- import org.springframework.boot.ApplicationRunner;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
- import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
- import org.springframework.data.redis.connection.RedisConnectionFactory;
- import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
- import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
- import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
- import org.springframework.data.redis.core.StringRedisTemplate;
- import org.springframework.lang.Nullable;
- import org.springframework.scheduling.annotation.Scheduled;
- import org.springframework.stereotype.Component;
- import javax.sql.DataSource;
- import java.sql.Connection;
- import java.time.Duration;
- import java.util.HashMap;
- import java.util.List;
- import java.util.Map;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- import java.util.concurrent.atomic.AtomicBoolean;
- /**
- * 腾讯-百度回传项目 Bean 组装配置类。
- * 负责所有核心 Bean 的创建和后台任务(Worker、Leader 选举、定时同步等)。
- */
- @Configuration
- public class AppConfiguration {
- private static final Logger log = LoggerFactory.getLogger(AppConfiguration.class);
- @Autowired private AppProperties props;
- // ─── Redis ───────────────────────────────────────────────────────────────
- @Bean
- public RedisConnectionFactory redisConnectionFactory() {
- String addr = props.getRedisAddr();
- String[] parts = addr.split(":", 2);
- String host = parts[0];
- int port = parts.length > 1 ? Integer.parseInt(parts[1]) : 6379;
- RedisStandaloneConfiguration cfg = new RedisStandaloneConfiguration(host, port);
- cfg.setDatabase(props.getRedisDb());
- if (props.getRedisUsername() != null && !props.getRedisUsername().isBlank()) {
- cfg.setUsername(props.getRedisUsername());
- }
- if (props.getRedisPassword() != null && !props.getRedisPassword().isBlank()) {
- cfg.setPassword(props.getRedisPassword());
- }
- LettuceClientConfiguration clientCfg = LettuceClientConfiguration.builder()
- .clientOptions(ClientOptions.builder()
- .protocolVersion(ProtocolVersion.RESP2)
- .socketOptions(SocketOptions.builder()
- .connectTimeout(props.getRedisConnectTimeout())
- .build())
- .build())
- .commandTimeout(props.getRedisCommandTimeout())
- .shutdownTimeout(props.getRedisShutdownTimeout())
- .build();
- LettuceConnectionFactory factory = new LettuceConnectionFactory(cfg, clientCfg);
- factory.setValidateConnection(props.isRedisValidateConnection());
- factory.setShareNativeConnection(false);
- return factory;
- }
- @Bean
- public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
- return new StringRedisTemplate(factory);
- }
- @Bean
- public RedisHotStore redisHotStore(StringRedisTemplate redis,
- ObjectMapper objectMapper,
- @Nullable TiDBColdStore coldStore) {
- return new RedisHotStore(redis, objectMapper,
- coldStore,
- "adx:tencent:",
- props.getRedisStream(),
- props.getBidTtl());
- }
- @Bean
- public RedisLock redisLock(StringRedisTemplate redis) {
- return new RedisLock(redis);
- }
- @Bean
- public LeaderElection leaderElection(RedisLock redisLock) {
- return new LeaderElection(redisLock);
- }
- // ─── TiDB(可选)────────────────────────────────────────────────────────
- @Bean
- public DataSource tidbDataSource() {
- String url = props.getTidbUrl();
- if (url == null || url.isBlank()) {
- log.info("TiDB URL is empty; cold storage disabled");
- return null;
- }
- try {
- return createDataSource("adx-tidb", url, props.getTidbUsername(), props.getTidbPassword());
- } catch (Exception e) {
- log.warn("Failed to open TiDB datasource: {}", e.getMessage(), e);
- return null;
- }
- }
- @Bean
- public TiDBColdStore tiDBColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
- if (tidbDataSource == null) {
- log.info("TiDB datasource is empty; cold storage disabled");
- return null;
- }
- try {
- SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
- TiDBColdStore store = new TiDBColdStore(sqlSessionFactory, objectMapper);
- store.migrate();
- log.info("TiDB cold store initialized and migrated");
- return store;
- } catch (Exception e) {
- log.warn("Failed to open TiDB: {}; cold storage disabled", e.getMessage(), e);
- return null;
- }
- }
- private SqlSessionFactory createSqlSessionFactory(DataSource dataSource) throws Exception {
- SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
- factoryBean.setDataSource(dataSource);
- factoryBean.setMapperLocations(
- new PathMatchingResourcePatternResolver().getResources("classpath:mapper/*.xml"));
- return factoryBean.getObject();
- }
- private DataSource createDataSource(String poolName, String jdbcUrl, String username, String password) {
- HikariConfig config = new HikariConfig();
- config.setDriverClassName("com.mysql.cj.jdbc.Driver");
- config.setJdbcUrl(jdbcUrl);
- if (username != null && !username.isBlank()) config.setUsername(username);
- if (password != null && !password.isBlank()) config.setPassword(password);
- config.setPoolName(poolName);
- config.setMinimumIdle(props.getTidbMinimumIdle());
- config.setMaximumPoolSize(props.getTidbMaximumPoolSize());
- config.setConnectionTimeout(props.getTidbConnectionTimeout().toMillis());
- config.setValidationTimeout(props.getTidbValidationTimeout().toMillis());
- config.setIdleTimeout(props.getTidbIdleTimeout().toMillis());
- config.setMaxLifetime(props.getTidbMaxLifetime().toMillis());
- config.setKeepaliveTime(props.getTidbKeepaliveTime().toMillis());
- return new HikariDataSource(config);
- }
- // ─── Baidu 客户端 ────────────────────────────────────────────────────────
- @Bean
- public AuctionPriceEncoder auctionPriceEncoder() {
- return AuctionPriceEncoder.create(
- props.getBaiduAuctionPriceEncryption(),
- props.getBaiduAuctionPriceEKey(),
- props.getBaiduAuctionPriceIKey());
- }
- @Bean
- public AdxClient adxClient(AuctionPriceEncoder encoder, ObjectMapper objectMapper) {
- return new AdxClient(props.getBaiduAdxEndpoint(), encoder, objectMapper);
- }
- @Bean
- public ConversionClient conversionClient(ObjectMapper objectMapper) {
- return new ConversionClient(
- props.getBaiduConversionBaseUrl(),
- props.getBaiduCustomerName(),
- props.getBaiduSecret(),
- objectMapper);
- }
- @Bean
- public HonorColdStore honorColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
- if (tidbDataSource == null) return null;
- try {
- SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
- HonorColdStore store = new HonorColdStore(sqlSessionFactory, objectMapper);
- store.migrate();
- return store;
- } catch (Exception e) {
- log.warn("Failed to open Honor TiDB: {}", e.getMessage(), e);
- return null;
- }
- }
- @Bean
- public HonorHotStore honorHotStore(StringRedisTemplate redis,
- ObjectMapper objectMapper,
- @Nullable HonorColdStore honorColdStore) {
- if (honorColdStore == null) return null;
- return new HonorHotStore(redis, objectMapper, honorColdStore,
- "adx:honor:", props.getHonorRedisStream(), props.getBidTtl());
- }
- @Bean
- public HonorPlacement honorPlacement() {
- HonorPlacement p = new HonorPlacement();
- p.setBaiduMediaId(props.getBaiduMediaId());
- p.setBaiduAppId(props.getBaiduAppId());
- p.setBaiduTagId(props.getBaiduTagId());
- p.setBidFloor(props.getBaiduBidFloor());
- p.setActionTypes(List.of(0, 1, 2));
- Map<String, HonorPlacement.HonorAppPlacement> platforms = new HashMap<>();
- String honorAndroidAppId = props.getBaiduAndroidAppId();
- List<String> honorAndroidTagIds = props.getBaiduAndroidTagIds();
- if (honorAndroidAppId != null && !honorAndroidAppId.isBlank()) {
- HonorPlacement.HonorAppPlacement android = new HonorPlacement.HonorAppPlacement();
- android.setAppId(honorAndroidAppId);
- android.setTagIds(honorAndroidTagIds);
- platforms.put("android", android);
- }
- String honorIosAppId = props.getBaiduIosAppId();
- List<String> honorIosTagIds = props.getBaiduIosTagIds();
- if (honorIosAppId != null && !honorIosAppId.isBlank()) {
- HonorPlacement.HonorAppPlacement ios = new HonorPlacement.HonorAppPlacement();
- ios.setAppId(honorIosAppId);
- ios.setTagIds(honorIosTagIds);
- platforms.put("ios", ios);
- }
- if (!platforms.isEmpty()) p.setPlatforms(platforms);
- return p;
- }
- @Bean
- public HonorClient honorClient(ObjectMapper objectMapper) {
- return new HonorClient(props.getHonorConversionBaseUrl(), props.getHonorChannelId(), objectMapper);
- }
- @Bean
- public HonorRetryService honorRetryService(@Nullable HonorColdStore honorColdStore, HonorClient honorClient) {
- if (honorColdStore == null) return null;
- return new HonorRetryService(honorColdStore, honorClient, props.getHonorCallbackRetryLimit());
- }
- @Bean
- public HonorColdWorker honorColdWorker(@Nullable HonorHotStore honorHotStore,
- @Nullable HonorColdStore honorColdStore,
- ObjectMapper objectMapper) {
- if (honorHotStore == null || honorColdStore == null) return null;
- return new HonorColdWorker(honorHotStore, honorColdStore,
- "honor-cold-writers", props.getWorkerConsumer() + "-honor", props.getWorkerBatch(),
- null, props.getRedisStreamMaxLen(), objectMapper);
- }
- @Bean
- public KuaishouColdStore kuaishouColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
- if (tidbDataSource == null) return null;
- try {
- SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
- KuaishouColdStore store = new KuaishouColdStore(sqlSessionFactory, objectMapper);
- store.migrate();
- return store;
- } catch (Exception e) {
- log.warn("Failed to open Kuaishou TiDB: {}", e.getMessage(), e);
- return null;
- }
- }
- @Bean
- public KuaishouHotStore kuaishouHotStore(StringRedisTemplate redis,
- ObjectMapper objectMapper,
- @Nullable KuaishouColdStore kuaishouColdStore) {
- if (kuaishouColdStore == null) return null;
- return new KuaishouHotStore(redis, objectMapper, kuaishouColdStore,
- "adx:kuaishou:", props.getKuaishouRedisStream(), props.getBidTtl());
- }
- @Bean
- public KuaishouPlacement kuaishouPlacement() {
- KuaishouPlacement p = new KuaishouPlacement();
- p.setBaiduMediaId(props.getBaiduMediaId());
- p.setBaiduAppId(props.getBaiduAppId());
- p.setBaiduTagId(props.getBaiduTagId());
- p.setBidFloor(props.getBaiduBidFloor());
- p.setActionTypes(List.of(0, 1, 2));
- Map<String, KuaishouPlacement.KuaishouAppPlacement> platforms = new HashMap<>();
- if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
- KuaishouPlacement.KuaishouAppPlacement android = new KuaishouPlacement.KuaishouAppPlacement();
- android.setAppId(props.getBaiduAndroidAppId());
- android.setTagIds(props.getBaiduAndroidTagIds());
- platforms.put("android", android);
- }
- if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
- KuaishouPlacement.KuaishouAppPlacement ios = new KuaishouPlacement.KuaishouAppPlacement();
- ios.setAppId(props.getBaiduIosAppId());
- ios.setTagIds(props.getBaiduIosTagIds());
- platforms.put("ios", ios);
- }
- if (!platforms.isEmpty()) p.setPlatforms(platforms);
- return p;
- }
- @Bean
- public KuaishouClient kuaishouClient(ObjectMapper objectMapper) {
- return new KuaishouClient(props.getKuaishouConversionBaseUrl(), objectMapper);
- }
- @Bean
- public KuaishouRetryService kuaishouRetryService(@Nullable KuaishouColdStore kuaishouColdStore,
- KuaishouClient kuaishouClient) {
- if (kuaishouColdStore == null) return null;
- return new KuaishouRetryService(kuaishouColdStore, kuaishouClient, props.getKuaishouCallbackRetryLimit());
- }
- @Bean
- public KuaishouColdWorker kuaishouColdWorker(@Nullable KuaishouHotStore kuaishouHotStore,
- @Nullable KuaishouColdStore kuaishouColdStore,
- ObjectMapper objectMapper) {
- if (kuaishouHotStore == null || kuaishouColdStore == null) return null;
- return new KuaishouColdWorker(kuaishouHotStore, kuaishouColdStore,
- "kuaishou-cold-writers", props.getWorkerConsumer() + "-kuaishou", props.getWorkerBatch(),
- null, props.getRedisStreamMaxLen(), objectMapper);
- }
- @Bean
- public VivoColdStore vivoColdStore(@Nullable DataSource tidbDataSource, ObjectMapper objectMapper) {
- if (tidbDataSource == null) return null;
- try {
- SqlSessionFactory sqlSessionFactory = createSqlSessionFactory(tidbDataSource);
- VivoColdStore store = new VivoColdStore(sqlSessionFactory, objectMapper);
- store.migrate();
- return store;
- } catch (Exception e) {
- log.warn("Failed to open Vivo TiDB: {}", e.getMessage(), e);
- return null;
- }
- }
- @Bean
- public VivoHotStore vivoHotStore(StringRedisTemplate redis,
- ObjectMapper objectMapper,
- @Nullable VivoColdStore vivoColdStore) {
- if (vivoColdStore == null) return null;
- return new VivoHotStore(redis, objectMapper, vivoColdStore,
- "adx:vivo:", props.getVivoRedisStream(), props.getBidTtl());
- }
- @Bean
- public VivoPlacement vivoPlacement() {
- VivoPlacement p = new VivoPlacement();
- p.setBaiduMediaId(props.getBaiduMediaId());
- p.setBaiduAppId(props.getBaiduAppId());
- p.setBaiduTagId(props.getBaiduTagId());
- p.setBidFloor(props.getBaiduBidFloor());
- p.setActionTypes(List.of(0, 1, 2));
- Map<String, VivoPlacement.VivoAppPlacement> platforms = new HashMap<>();
- if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
- VivoPlacement.VivoAppPlacement android = new VivoPlacement.VivoAppPlacement();
- android.setAppId(props.getBaiduAndroidAppId());
- android.setTagIds(props.getBaiduAndroidTagIds());
- platforms.put("android", android);
- }
- if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
- VivoPlacement.VivoAppPlacement ios = new VivoPlacement.VivoAppPlacement();
- ios.setAppId(props.getBaiduIosAppId());
- ios.setTagIds(props.getBaiduIosTagIds());
- platforms.put("ios", ios);
- }
- if (!platforms.isEmpty()) p.setPlatforms(platforms);
- return p;
- }
- @Bean
- public VivoAuthService vivoAuthService(StringRedisTemplate redisTemplate,
- @Nullable VivoColdStore vivoColdStore,
- ObjectMapper objectMapper) {
- if (vivoColdStore == null) return null;
- return new VivoAuthService(
- props.getVivoConversionBaseUrl(),
- props.getVivoRedirectUri(),
- props.getVivoTokenRedisPrefix(),
- props.getVivoAdvertiserRedisPrefix(),
- redisTemplate,
- vivoColdStore,
- objectMapper
- );
- }
- @Bean
- public VivoClient vivoClient(@Nullable VivoAuthService vivoAuthService, ObjectMapper objectMapper) {
- if (vivoAuthService == null) return null;
- return new VivoClient(props.getVivoConversionBaseUrl(), vivoAuthService, objectMapper);
- }
- @Bean
- public VivoRetryService vivoRetryService(@Nullable VivoColdStore vivoColdStore,
- @Nullable VivoClient vivoClient) {
- if (vivoColdStore == null || vivoClient == null) return null;
- return new VivoRetryService(vivoColdStore, vivoClient, props.getVivoCallbackRetryLimit());
- }
- @Bean
- public VivoAdvertiserSyncService vivoAdvertiserSyncService(@Nullable VivoAuthService vivoAuthService,
- @Nullable VivoColdStore vivoColdStore,
- ObjectMapper objectMapper) {
- if (vivoAuthService == null || vivoColdStore == null) return null;
- return new VivoAdvertiserSyncService(props.getVivoConversionBaseUrl(), vivoAuthService, vivoColdStore, objectMapper);
- }
- @Bean
- public VivoColdWorker vivoColdWorker(@Nullable VivoHotStore vivoHotStore,
- @Nullable VivoColdStore vivoColdStore,
- ObjectMapper objectMapper) {
- if (vivoHotStore == null || vivoColdStore == null) return null;
- return new VivoColdWorker(vivoHotStore, vivoColdStore,
- "vivo-cold-writers", props.getWorkerConsumer() + "-vivo", props.getWorkerBatch(),
- null, props.getRedisStreamMaxLen(), objectMapper);
- }
- // ─── Tencent 客户端 + Sync 服务 ──────────────────────────────────────────
- @Bean
- public TencentClient tencentClient(ObjectMapper objectMapper, StringRedisTemplate redisTemplate) {
- // token 从 Redis 动态获取,支持外部刷新
- String tokenRedisKey = props.getTencentAccessTokenRedisKey();
- java.util.function.Supplier<String> tokenSupplier;
- if (tokenRedisKey != null && !tokenRedisKey.isBlank()) {
- tokenSupplier = () -> redisTemplate.opsForValue().get(tokenRedisKey);
- } else {
- // 兜底:使用配置文件中的静态 token
- String staticToken = props.getTencentAccessToken();
- tokenSupplier = () -> staticToken;
- }
- return new TencentClient(
- tokenSupplier,
- props.getTencentActMap(),
- objectMapper);
- }
- @Bean
- public TagEventResolver tagEventResolver(@Nullable TiDBColdStore coldStore,
- StringRedisTemplate redisTemplate) {
- if (coldStore == null) return null;
- return new TagEventResolver(redisTemplate, coldStore, props.getTagEventSyncRedisPrefix());
- }
- @Bean
- public HonorTagEventResolver honorTagEventResolver(@Nullable HonorColdStore honorColdStore,
- StringRedisTemplate redisTemplate) {
- if (honorColdStore == null) return null;
- return new HonorTagEventResolver(redisTemplate, honorColdStore, props.getHonorTagEventSyncRedisPrefix());
- }
- @Bean
- public KuaishouTagEventResolver kuaishouTagEventResolver(@Nullable KuaishouColdStore kuaishouColdStore,
- StringRedisTemplate redisTemplate) {
- if (kuaishouColdStore == null) return null;
- return new KuaishouTagEventResolver(redisTemplate, kuaishouColdStore, props.getKuaishouTagEventSyncRedisPrefix());
- }
- @Bean
- public VivoTagEventResolver vivoTagEventResolver(@Nullable VivoColdStore vivoColdStore,
- StringRedisTemplate redisTemplate) {
- if (vivoColdStore == null) return null;
- return new VivoTagEventResolver(redisTemplate, vivoColdStore, props.getVivoTagEventSyncRedisPrefix());
- }
- @Bean
- public ConversionSyncService conversionSyncService(ConversionClient baiduClient,
- RedisHotStore hotStore,
- TencentClient tencentClient,
- @Nullable TiDBColdStore coldStore,
- @Nullable TagEventResolver tagEventResolver,
- @Nullable HonorTagEventResolver honorTagEventResolver,
- @Nullable KuaishouTagEventResolver kuaishouTagEventResolver,
- @Nullable VivoTagEventResolver vivoTagEventResolver,
- @Nullable HonorHotStore honorHotStore,
- @Nullable HonorColdStore honorColdStore,
- @Nullable KuaishouHotStore kuaishouHotStore,
- @Nullable KuaishouColdStore kuaishouColdStore,
- @Nullable VivoHotStore vivoHotStore,
- @Nullable VivoColdStore vivoColdStore,
- HonorClient honorClient,
- KuaishouClient kuaishouClient,
- @Nullable VivoClient vivoClient) {
- if (coldStore == null) return null;
- return new ConversionSyncService(baiduClient, hotStore, tencentClient, coldStore, tagEventResolver,
- honorTagEventResolver, kuaishouTagEventResolver, vivoTagEventResolver,
- honorHotStore, honorColdStore, kuaishouHotStore, kuaishouColdStore, vivoHotStore, vivoColdStore,
- honorClient, kuaishouClient, vivoClient, props);
- }
- @Bean
- public ConversionSyncRunner conversionSyncRunner(@Nullable ConversionSyncService syncer) {
- if (syncer == null) return null;
- return new ConversionSyncRunner(syncer,
- props.getConversionSyncDateOffsetDays(),
- props.getConversionSyncPageSize(),
- props.getConversionSyncActs());
- }
- @Bean
- public ConversionBackfillJobService conversionBackfillJobService(@Nullable ConversionSyncRunner syncer) {
- if (syncer == null) return null;
- return new ConversionBackfillJobService(syncer);
- }
- @Bean
- public RetryService retryService(@Nullable TiDBColdStore coldStore, TencentClient tencentClient) {
- if (coldStore == null) return null;
- return new RetryService(coldStore, tencentClient, props.getCallbackRetryLimit());
- }
- @Bean
- public ManualTencentCallbackService manualTencentCallbackService(@Nullable TiDBColdStore coldStore,
- RedisHotStore hotStore,
- TencentClient tencentClient) {
- if (coldStore == null) return null;
- return new ManualTencentCallbackService(coldStore, hotStore, tencentClient);
- }
- @Bean
- public TagEventSyncService tagEventSyncService(@Nullable TiDBColdStore coldStore,
- StringRedisTemplate redisTemplate) {
- if (coldStore == null) return null;
- // TTL 取同步间隔的 3 倍,避免表中已删除的 tag 在 Redis 长期残留
- Duration ttl = props.getTagEventSyncInterval().multipliedBy(3);
- return new TagEventSyncService(coldStore, redisTemplate,
- props.getTagEventSyncRedisPrefix(), ttl);
- }
- @Bean
- public HonorTagEventSyncService honorTagEventSyncService(@Nullable HonorColdStore honorColdStore,
- StringRedisTemplate redisTemplate) {
- if (honorColdStore == null) return null;
- Duration ttl = props.getHonorTagEventSyncInterval().multipliedBy(3);
- return new HonorTagEventSyncService(honorColdStore, redisTemplate,
- props.getHonorTagEventSyncRedisPrefix(), ttl);
- }
- @Bean
- public KuaishouTagEventSyncService kuaishouTagEventSyncService(@Nullable KuaishouColdStore kuaishouColdStore,
- StringRedisTemplate redisTemplate) {
- if (kuaishouColdStore == null) return null;
- Duration ttl = props.getKuaishouTagEventSyncInterval().multipliedBy(3);
- return new KuaishouTagEventSyncService(kuaishouColdStore, redisTemplate,
- props.getKuaishouTagEventSyncRedisPrefix(), ttl);
- }
- @Bean
- public VivoTagEventSyncService vivoTagEventSyncService(@Nullable VivoColdStore vivoColdStore,
- StringRedisTemplate redisTemplate) {
- if (vivoColdStore == null) return null;
- Duration ttl = props.getVivoTagEventSyncInterval().multipliedBy(3);
- return new VivoTagEventSyncService(vivoColdStore, redisTemplate,
- props.getVivoTagEventSyncRedisPrefix(), ttl);
- }
- @Bean
- public AdBidReportStore adBidReportStore(@Nullable DataSource tidbDataSource) {
- if (tidbDataSource == null) return null;
- String mainSchema = resolveDatabaseName(tidbDataSource, props.getTidbUrl());
- if (mainSchema == null || mainSchema.isBlank()) return null;
- AdBidReportStore store = new AdBidReportStore(
- tidbDataSource,
- mainSchema,
- props.getReportDatabaseName(),
- props.getAdBidReportRetentionDays(),
- props.getAdBidReportArchiveBatchSize(),
- props.getAdBidReportArchiveMaxBatchesPerRun(),
- props.getAdBidReportStatsLookbackDays());
- store.migrate();
- return store;
- }
- // ─── ColdWorker ─────────────────────────────────────────────────────────
- @Bean
- public ColdWorker coldWorker(RedisHotStore hotStore, @Nullable TiDBColdStore coldStore,
- ObjectMapper objectMapper) {
- if (coldStore == null) return null;
- return new ColdWorker(hotStore, coldStore,
- props.getWorkerGroup(), props.getWorkerConsumer(),
- props.getWorkerBatch(), null,
- props.getRedisStreamMaxLen(), objectMapper);
- }
- // ─── MediaPlacement(腾讯)───────────────────────────────────────────────
- @Bean
- public MediaPlacement tencentPlacement() {
- MediaPlacement p = new MediaPlacement();
- p.setBaiduMediaId(props.getBaiduMediaId());
- p.setBaiduAppId(props.getBaiduAppId());
- p.setBaiduTagId(props.getBaiduTagId());
- p.setBidFloor(props.getBaiduBidFloor());
- p.setActionTypes(List.of(0, 1, 2));
- Map<String, MediaPlacement.BaiduAppPlacement> platforms = new HashMap<>();
- if (props.getBaiduAndroidAppId() != null && !props.getBaiduAndroidAppId().isBlank()) {
- MediaPlacement.BaiduAppPlacement android = new MediaPlacement.BaiduAppPlacement();
- android.setAppId(props.getBaiduAndroidAppId());
- android.setTagIds(props.getBaiduAndroidTagIds());
- platforms.put("android", android);
- }
- if (props.getBaiduIosAppId() != null && !props.getBaiduIosAppId().isBlank()) {
- MediaPlacement.BaiduAppPlacement ios = new MediaPlacement.BaiduAppPlacement();
- ios.setAppId(props.getBaiduIosAppId());
- ios.setTagIds(props.getBaiduIosTagIds());
- platforms.put("ios", ios);
- }
- if (!platforms.isEmpty()) p.setPlatforms(platforms);
- return p;
- }
- private static String resolveDatabaseName(DataSource dataSource, String jdbcUrl) {
- try (Connection connection = dataSource.getConnection()) {
- String catalog = connection.getCatalog();
- if (catalog != null && !catalog.isBlank()) {
- return catalog.trim();
- }
- } catch (Exception ignored) {
- }
- return extractDatabaseName(jdbcUrl);
- }
- private static String extractDatabaseName(String jdbcUrl) {
- if (jdbcUrl == null || jdbcUrl.isBlank()) {
- return null;
- }
- String value = jdbcUrl.trim();
- int scheme = value.indexOf("://");
- int searchFrom = scheme >= 0 ? scheme + 3 : 0;
- int slash = value.indexOf('/', searchFrom);
- if (slash < 0 || slash + 1 >= value.length()) {
- return null;
- }
- int end = value.length();
- int question = value.indexOf('?', slash + 1);
- if (question >= 0) {
- end = Math.min(end, question);
- }
- int hash = value.indexOf('#', slash + 1);
- if (hash >= 0) {
- end = Math.min(end, hash);
- }
- int semicolon = value.indexOf(';', slash + 1);
- if (semicolon >= 0) {
- end = Math.min(end, semicolon);
- }
- int nextSlash = value.indexOf('/', slash + 1);
- if (nextSlash >= 0 && nextSlash < end) {
- end = nextSlash;
- }
- String db = value.substring(slash + 1, end);
- return db == null || db.isBlank() ? null : db.trim();
- }
- // ─── 后台任务调度 ────────────────────────────────────────────────────────
- @Component
- static class BackgroundTasks implements ApplicationRunner {
- private static final Logger taskLog = LoggerFactory.getLogger(BackgroundTasks.class);
- @Autowired private AppProperties props;
- @Autowired(required = false) private ColdWorker coldWorker;
- @Autowired(required = false) private ConversionSyncRunner conversionSyncRunner;
- @Autowired(required = false) private RetryService retryService;
- @Autowired(required = false) private TagEventSyncService tagEventSyncService;
- @Autowired(required = false) private LeaderElection leaderElection;
- private final AtomicBoolean stopped = new AtomicBoolean(false);
- private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
- Thread t = new Thread(r);
- t.setDaemon(true);
- return t;
- });
- @Override
- public void run(ApplicationArguments args) {
- // 冷路径 Worker
- if (coldWorker != null) {
- executor.submit(() -> {
- while (!stopped.get()) {
- try {
- coldWorker.runOnce();
- } catch (Exception e) {
- taskLog.error("cold worker: {}", e.getMessage(), e);
- sleep(1000);
- }
- }
- });
- taskLog.info("Cold worker started");
- }
- // 转化同步
- if (conversionSyncRunner != null && props.isConversionSyncEnabled()) {
- if (props.isSkipLeaderElection()) {
- executor.submit(() -> runConversionSync(() -> stopped.get()));
- taskLog.info("Conversion sync started (skip leader election)");
- } else if (leaderElection != null) {
- executor.submit(() ->
- leaderElection.run(
- "adx:lock:tencent:conversion-sync",
- props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
- stopped::get,
- (jobStop) -> runConversionSync(jobStop),
- e -> taskLog.error("tencent conversion sync leader: {}", e.getMessage(), e)
- )
- );
- taskLog.info("Conversion sync leader election started");
- }
- } else {
- taskLog.warn("Conversion sync NOT started: runner={}, enabled={}",
- conversionSyncRunner != null, props.isConversionSyncEnabled());
- }
- // 回调重试
- if (retryService != null && props.isCallbackRetryEnabled()) {
- if (props.isSkipLeaderElection()) {
- executor.submit(() -> runCallbackRetry(() -> stopped.get()));
- taskLog.info("Callback retry started (skip leader election)");
- } else if (leaderElection != null) {
- executor.submit(() ->
- leaderElection.run(
- "adx:lock:tencent:callback-retry",
- props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
- stopped::get,
- (jobStop) -> runCallbackRetry(jobStop),
- e -> taskLog.error("tencent callback retry leader: {}", e.getMessage(), e)
- )
- );
- taskLog.info("Callback retry leader election started");
- }
- } else {
- taskLog.warn("Callback retry NOT started: retryService={}, enabled={}",
- retryService != null, props.isCallbackRetryEnabled());
- }
- // 广告位回传方式同步
- if (tagEventSyncService != null && props.isTagEventSyncEnabled()) {
- if (props.isSkipLeaderElection()) {
- executor.submit(() -> runTagEventSync(() -> stopped.get()));
- taskLog.info("Tag event sync started (skip leader election)");
- } else if (leaderElection != null) {
- executor.submit(() ->
- leaderElection.run(
- "adx:lock:tencent:tag-event-sync",
- props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
- stopped::get,
- (jobStop) -> runTagEventSync(jobStop),
- e -> taskLog.error("tencent tag event sync leader: {}", e.getMessage(), e)
- )
- );
- taskLog.info("Tag event sync leader election started");
- }
- } else {
- taskLog.warn("Tag event sync NOT started: service={}, enabled={}",
- tagEventSyncService != null, props.isTagEventSyncEnabled());
- }
- }
- private void runConversionSync(LeaderElection.StopSignal jobStop) {
- long intervalMs = props.getConversionSyncInterval().toMillis();
- taskLog.info("[ConversionSync] task started, interval={}ms", intervalMs);
- while (!jobStop.isStopped()) {
- try {
- taskLog.info("[ConversionSync] executing...");
- ConversionSyncService.SyncResult r = conversionSyncRunner.runOnce();
- taskLog.info("[ConversionSync] done: fetched={} sent={} failed={}", r.fetched.get(), r.sent.get(), r.failed.get());
- } catch (Exception e) {
- taskLog.error("[ConversionSync] error: {}", e.getMessage(), e);
- }
- taskLog.info("[ConversionSync] sleeping {}ms until next run", intervalMs);
- sleepResponsive(intervalMs, jobStop);
- }
- taskLog.info("[ConversionSync] task stopped");
- }
- @Scheduled(cron = "0 17 1,7,9,11,16,20,23 * * ?")
- public void runDelayedConversionSync() {
- if (conversionSyncRunner == null || !props.isConversionSyncEnabled()) {
- return;
- }
- if (stopped.get()) {
- return;
- }
- Runnable job = () -> {
- long startNs = System.nanoTime();
- try {
- taskLog.info("[ConversionBackfill] executing for offsets=1..5");
- ConversionSyncService.SyncResult r = conversionSyncRunner.runBackfillForOffsetsParallel(
- List.of(-1, -2, -3, -4, -5), 3);
- long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
- taskLog.info("[ConversionBackfill] done: fetched={} matched={} sent={} failed={} skipped={} alreadySent={}",
- r.fetched.get(), r.matched.get(), r.sent.get(), r.failed.get(),
- r.skipped.get(), r.alreadySent.get());
- taskLog.info("[ConversionBackfill] total cost={}ms", costMs);
- } catch (Exception e) {
- long costMs = java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
- taskLog.error("[ConversionBackfill] error after {}ms: {}", costMs, e.getMessage(), e);
- }
- };
- if (props.isSkipLeaderElection() || leaderElection == null) {
- job.run();
- return;
- }
- executor.submit(() ->
- leaderElection.runOnce(
- "adx:lock:tencent:conversion-backfill",
- props.getTaskLockTtl(), props.getTaskLockRenewInterval(),
- stopped::get,
- stop -> job.run(),
- e -> taskLog.error("tencent conversion backfill leader: {}", e.getMessage(), e)
- )
- );
- }
- private void runCallbackRetry(LeaderElection.StopSignal jobStop) {
- long intervalMs = props.getCallbackRetryInterval().toMillis();
- taskLog.info("[CallbackRetry] task started, interval={}ms, limit={}", intervalMs, props.getCallbackRetryLimit());
- while (!jobStop.isStopped()) {
- try {
- taskLog.info("[CallbackRetry] executing...");
- RetryService.RetryResult r = retryService.retryTencentCallbacks(props.getCallbackRetryLimit());
- taskLog.info("[CallbackRetry] done: fetched={} sent={} failed={}", r.fetched, r.sent, r.failed);
- } catch (Exception e) {
- taskLog.error("[CallbackRetry] error: {}", e.getMessage(), e);
- }
- taskLog.info("[CallbackRetry] sleeping {}ms until next run", intervalMs);
- sleepResponsive(intervalMs, jobStop);
- }
- taskLog.info("[CallbackRetry] task stopped");
- }
- private void runTagEventSync(LeaderElection.StopSignal jobStop) {
- long intervalMs = props.getTagEventSyncInterval().toMillis();
- taskLog.info("[TagEventSync] task started, interval={}ms", intervalMs);
- while (!jobStop.isStopped()) {
- try {
- int n = tagEventSyncService.syncOnce();
- taskLog.info("[TagEventSync] done: synced={}", n);
- } catch (Exception e) {
- taskLog.error("[TagEventSync] error: {}", e.getMessage(), e);
- }
- sleepResponsive(intervalMs, jobStop);
- }
- taskLog.info("[TagEventSync] task stopped");
- }
- private static void sleepResponsive(long ms, LeaderElection.StopSignal jobStop) {
- long deadline = System.currentTimeMillis() + ms;
- while (!jobStop.isStopped()) {
- long remaining = deadline - System.currentTimeMillis();
- if (remaining <= 0) break;
- try {
- Thread.sleep(Math.min(remaining, 200));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- break;
- }
- }
- }
- private static void sleep(long ms) {
- try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
- }
- }
- }
|