|
|
@@ -0,0 +1,197 @@
|
|
|
+package com.webflux.launchadmin.mysql.task;
|
|
|
+
|
|
|
+import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
+import com.fasterxml.jackson.databind.SerializationFeature;
|
|
|
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
|
|
+import com.plumelog.core.util.LogExceptionStackTrace;
|
|
|
+import com.webflux.launchadmin.global.BaseException;
|
|
|
+import com.fasterxml.jackson.databind.module.SimpleModule;
|
|
|
+import com.webflux.launchadmin.mysql.entity.planNew.PlanNew;
|
|
|
+import com.weblux.launchredis.utils.FissionRedisKey;
|
|
|
+import jakarta.annotation.Resource;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
|
|
|
+import org.jetbrains.annotations.NotNull;
|
|
|
+import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
|
|
|
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
|
|
|
+import org.springframework.data.relational.core.query.Criteria;
|
|
|
+import org.springframework.data.relational.core.query.Query;
|
|
|
+import org.springframework.scheduling.annotation.Scheduled;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+import reactor.core.publisher.Flux;
|
|
|
+import reactor.core.publisher.Mono;
|
|
|
+import com.fasterxml.jackson.core.JsonParser;
|
|
|
+import com.fasterxml.jackson.databind.DeserializationContext;
|
|
|
+import com.fasterxml.jackson.databind.JsonDeserializer;
|
|
|
+
|
|
|
+import java.io.IOException;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.time.ZonedDateTime;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.util.List;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+@Component
|
|
|
+@Slf4j
|
|
|
+public class SyncPlanNewToRedis {
|
|
|
+ @Resource
|
|
|
+ private R2dbcEntityTemplate template;
|
|
|
+ @Resource
|
|
|
+ private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * SELECT GROUP_CONCAT(COLUMN_NAME SEPARATOR ', ') AS column_names
|
|
|
+ * FROM information_schema.COLUMNS
|
|
|
+ * WHERE TABLE_NAME = 'plan_new' AND TABLE_SCHEMA = 'wx_share_test';
|
|
|
+ */
|
|
|
+ // String column="plan_new_id, name, type, code, group_id, " +
|
|
|
+ // "poster_template_id, page_show, page_line, count, list_audio_id, ret_audio_id, link, is_add_qun, " +
|
|
|
+ // "background_group_id, home_page_id, top_adv_num, bottom_adv_num, return_loop_id, background_group_qs_id, " +
|
|
|
+ // "top_adv_bl, bottom_adv_bl, return_loop_bl, group_Id_qs, baidu_dot, background_group_id_new_add, preview_url," +
|
|
|
+ // " redirect_type, redirect_plan_id, import_domain, landing_page_domain, transfer_page_domain, transfer_page_bl," +
|
|
|
+ // " backgroup_domain, list_button, group_list_display";
|
|
|
+ /**
|
|
|
+ *同步裂变计划
|
|
|
+ */
|
|
|
+ @Scheduled(cron="0 */1 * * * ?")
|
|
|
+ @SchedulerLock(name = "launch:SyncFissionPlan:syncPlanNew", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
|
|
|
+ public void syncFissionPlan(){
|
|
|
+ Mono<List<PlanNew>> listMono = template.select(PlanNew.class).all().collectList();
|
|
|
+ listMono.flatMap(list -> {
|
|
|
+ List<String> collect = list.stream().map(PlanNew::code).collect(Collectors.toList());
|
|
|
+ Mono<List<String>> listMono1 = reactiveRedisTemplate
|
|
|
+ .opsForSet()
|
|
|
+ .members(FissionRedisKey.PLAN_NEW_INFO_SET).collectList()
|
|
|
+ .switchIfEmpty(Mono.just(List.of()));
|
|
|
+ return Mono.zip(Mono.just(collect),listMono1,Mono.just(list)).flatMap(tupleN->{
|
|
|
+ var mids = tupleN.getT1();
|
|
|
+ var rids = tupleN.getT2();
|
|
|
+ var planNews = tupleN.getT3();
|
|
|
+ // +
|
|
|
+ List<String> addIds = mids.stream().filter(mid -> !rids.contains(mid)).collect(Collectors.toList());
|
|
|
+ Mono<Void> addMono=Mono.empty();
|
|
|
+ if (!addIds.isEmpty()) {
|
|
|
+ List<PlanNew> addPlanNew = planNews.stream().filter(f -> addIds.contains(f.code())).collect(Collectors.toList());
|
|
|
+ if (!addPlanNew.isEmpty()) {
|
|
|
+ addMono = Flux.fromIterable(addPlanNew)
|
|
|
+ .flatMap(this::setRedisValue).then();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // -
|
|
|
+ Mono<Void> then1 =Mono.empty();
|
|
|
+ List<String> removeIds = rids.stream().filter(rid -> !mids.contains(rid)).collect(Collectors.toList());
|
|
|
+ if (!removeIds.isEmpty()) {
|
|
|
+ then1 = Flux.fromIterable(removeIds).flatMap(this::getDelete).then();
|
|
|
+ }
|
|
|
+ return Mono.zip(addMono,then1).then();
|
|
|
+ }).then();
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ @NotNull
|
|
|
+ public Mono<Void> setRedisValue(PlanNew planNew) {
|
|
|
+ Mono<Long> add = reactiveRedisTemplate.opsForSet().add(FissionRedisKey.PLAN_NEW_INFO_SET, planNew.code());
|
|
|
+ String json= getPlanNewToString(planNew);
|
|
|
+ Mono<Boolean> set = reactiveRedisTemplate
|
|
|
+ .opsForValue()
|
|
|
+ .set(FissionRedisKey.PLAN_NEW_INFO_KEY + planNew.code(), json);
|
|
|
+ return set.then(add).then();
|
|
|
+ }
|
|
|
+
|
|
|
+ private String getPlanNewToString(PlanNew planNew) {
|
|
|
+ ObjectMapper objectMapper = new ObjectMapper();
|
|
|
+ objectMapper.registerModule(new JavaTimeModule());
|
|
|
+ objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
|
|
|
+ try {
|
|
|
+ // 将 record 转换为 JSON 字符串
|
|
|
+ return objectMapper.writeValueAsString(planNew);
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
+ throw new BaseException("PlanNew 将 record 转换为 JSON 字符串 异常：：："+e.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ @NotNull
|
|
|
+ public Mono<Void> getDelete(String code) {
|
|
|
+ Mono<Long> remove = reactiveRedisTemplate.opsForSet().remove(FissionRedisKey.PLAN_NEW_INFO_SET, code);
|
|
|
+ Mono<Boolean> delete = reactiveRedisTemplate.opsForValue().delete(FissionRedisKey.PLAN_NEW_INFO_KEY + code);
|
|
|
+ return remove.then(delete).then();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 通过code从数据库获取PlanNew并缓存redis
|
|
|
+ * @param code
|
|
|
+ * @return
|
|
|
+ */
|
|
|
+ public Mono<PlanNew> byCodeAndMysqlGetPlanNew(String code){
|
|
|
+ Mono<PlanNew> planNewMono = template
|
|
|
+ .selectOne(Query.query(Criteria.where("code").is(code)), PlanNew.class)//.columns(column.split(","))
|
|
|
+ .switchIfEmpty(Mono.error(new BaseException("计划code异常 返回结果为空"))) //Mono.error(new BaseException("查询海报模版 id:"+f.posterTemplateId()+"=>空"))
|
|
|
+ .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())));
|
|
|
+ return planNewMono.flatMap(planNew -> {
|
|
|
+ Mono<Void> voidMono = setRedisValue(planNew);
|
|
|
+ return voidMono.then(Mono.just(planNew));
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 通过redis获取缓存 空或失败 从mysql中取
|
|
|
+ * @param code
|
|
|
+ * @return
|
|
|
+ */
|
|
|
+ public Mono<PlanNew> byRedisGetPlanNewIfErByMysql(String code){
|
|
|
+ return reactiveRedisTemplate.opsForValue()
|
|
|
+ .get(FissionRedisKey.PLAN_NEW_INFO_KEY + code)
|
|
|
+ .switchIfEmpty(Mono.just(""))
|
|
|
+ .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())))
|
|
|
+ .flatMap(s -> {
|
|
|
+ Mono<PlanNew> planNewMono;
|
|
|
+ if(!"".equals(s)){
|
|
|
+ try {
|
|
|
+ PlanNew planNew = strParsePlanNew(s);
|
|
|
+ planNewMono=Mono.just(planNew);
|
|
|
+ }catch (Exception e){
|
|
|
+ log.error("接口异常 {} e = {}","获取redis Value 解析PlanNew失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e));
|
|
|
+ planNewMono = byCodeAndMysqlGetPlanNew(code);
|
|
|
+ }
|
|
|
+ }else {
|
|
|
+ planNewMono = byCodeAndMysqlGetPlanNew(code);
|
|
|
+ }
|
|
|
+ return planNewMono;
|
|
|
+ });
|
|
|
+ }
|
|
|
+ /**
|
|
|
+ * str 转PlanNew
|
|
|
+ * @param str
|
|
|
+ * @return
|
|
|
+ */
|
|
|
+ public PlanNew strParsePlanNew(String str) {
|
|
|
+ // 创建 ObjectMapper 实例并配置
|
|
|
+ ObjectMapper objectMapper = new ObjectMapper();
|
|
|
+ objectMapper.registerModule(new JavaTimeModule());
|
|
|
+ SimpleModule module = new SimpleModule();
|
|
|
+ module.addDeserializer(ZonedDateTime.class, new CustomZonedDateTimeDeserializer());
|
|
|
+ objectMapper.registerModule(module);
|
|
|
+ try {
|
|
|
+ // 将 JSON 字符串转换为 record
|
|
|
+ return objectMapper.readValue(str, PlanNew.class);
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
+ throw new BaseException(" 将 JSON字符串 转换为 PlanNew 异常：：："+e.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 自定义反序列化器
|
|
|
+ public static class CustomZonedDateTimeDeserializer extends JsonDeserializer<ZonedDateTime> {
|
|
|
+ private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+ @Override
|
|
|
+ public ZonedDateTime deserialize(JsonParser p, DeserializationContext ctxt)
|
|
|
+ throws IOException, JsonProcessingException {
|
|
|
+ String date = p.getText();
|
|
|
+ LocalDateTime localDateTime = LocalDateTime.parse(date, FORMATTER);
|
|
|
+ // 使用系统默认时区，可以根据需要修改
|
|
|
+ return ZonedDateTime.of(localDateTime, ZoneId.systemDefault());
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
MOKASZ\lw12420