裂变计划redis v0.07 +听书 +其他

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncAudioToRedis.java

@@ -0,0 +1,187 @@
+package com.webflux.launchadmin.mysql.task;
+
+import cn.hutool.core.util.StrUtil;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.plumelog.core.util.LogExceptionStackTrace;
+import com.webflux.launchadmin.global.BaseException;
+import com.webflux.launchadmin.mysql.entity.planNew.Audio;
+import com.webflux.launchadmin.mysql.entity.planNew.ReturnLoop;
+import com.weblux.launchredis.utils.AudioRedisKey;
+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.domain.Sort;
+import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.core.ScanOptions;
+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 java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
+public class SyncAudioToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+
+    private static final String COLUMN="id,url,name,type";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+//    template.selectOne(Query.query(Criteria.where("id").is(id)), Audio.class)
+//            .switchIfEmpty(Mono.error(new BaseException(StrUtil.format("通过计划retAudioId：{}查询音频为空", id))))
+//            .onErrorResume(throwable -> Mono.error(new BaseException(StrUtil.format("通过计划retAudioId：{}查询音频Error::{}", id, throwable.getMessage()))));
+
+    @Scheduled(fixedDelay = 130000)
+    @SchedulerLock(name = "launch:SyncAudioToRedis:syncAudioPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
+    public void syncAudioPlan(){
+        Mono<List<Audio>> listMono = template
+                .select(Query.query(Criteria.where("deleted_at").isNull())
+                        .columns(COLUMNARRAY)
+                , Audio.class).collectList();
+        listMono.flatMap(list -> {
+            if (!list.isEmpty()) {
+                Flux.fromIterable(list).flatMap(this::setRedisValue).subscribe();
+            }
+            return Mono.empty();
+        }).subscribe();
+        Mono<List<Audio>> listMonoDelette = template
+                .select(Query.query(Criteria.where("deleted_at").isNotNull())
+                                .columns(COLUMNARRAY)
+                        , Audio.class).collectList();
+        listMonoDelette.flatMap(list -> {
+            if (!list.isEmpty()) {
+                Flux.fromIterable(list).flatMap(this::deleteRedis).subscribe();
+            }
+            return Mono.empty();
+        }).subscribe();
+    }
+
+    @NotNull
+    public Mono<Void> setRedisValue(Audio audio) {
+        String json= getPlanNewToString(audio);
+        reactiveRedisTemplate.opsForHash().put(AudioRedisKey.AUDIO_INFO_HASH , String.valueOf(audio.id()), json).subscribe();
+       return Mono.empty();
+    }
+
+    private String getPlanNewToString(Audio 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());
+        }
+    }
+
+    public Mono<Void> deleteRedis(Audio audio) {
+        reactiveRedisTemplate.opsForHash().remove(AudioRedisKey.AUDIO_INFO_HASH, String.valueOf(audio.id())).subscribe();
+        return Mono.empty();
+    }
+
+    /**
+     * 通过pageShow从数据库获取 并缓存redis
+     * @param id
+     * @return
+     */
+    public Mono<Audio> byIdFromMysqlGetAudio(Long id){
+        Mono<Audio> id1 = template.selectOne(Query.query(Criteria.where("id").is(id)), Audio.class)
+                .switchIfEmpty(Mono.error(new BaseException(StrUtil.format("通过计划retAudioId：{}查询音频为空", id))))
+                .onErrorResume(throwable ->
+                        Mono.error(new BaseException(StrUtil.format("通过计划retAudioId：{}查询音频Error::{}", id, throwable.getMessage()))));
+        return id1.flatMap(audio -> {
+            setRedisValue(audio).subscribe();
+           return Mono.just(audio);
+        });
+    }
+
+    /**
+     * 通过redis获取缓存 空或失败 从mysql中取
+     * @param
+     * @return
+     */
+    public Mono<Audio> byRedisGetAudioIfErByMysql(Long id){
+      return   reactiveRedisTemplate
+              .opsForHash()
+              .get(AudioRedisKey.AUDIO_INFO_HASH,id)
+              .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis Audio 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return Mono.just("");
+              })
+              .flatMap(s -> {
+                  Mono<Audio> planNewMono;
+                  if(!"".equals(s) && Objects.nonNull(s)){
+                      try {
+                          Audio planNew = strParsePlanNew(String.valueOf(s));
+                          planNewMono=Mono.just(planNew);
+                      }catch (Exception e){
+                          log.error("接口异常  {} e = {} id={}","获取redis Value 解析Audio失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e),id);
+                          planNewMono = byIdFromMysqlGetAudio(id);
+                      }
+                  }else {
+                      planNewMono = byIdFromMysqlGetAudio(id);
+                  }
+                  return planNewMono;
+         });
+    }
+    /**
+     * str 解析对象
+     * @param str
+     * @return
+     */
+    public Audio 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, Audio.class);
+            } catch (JsonProcessingException e) {
+                String format = StrUtil.format(" 将 JSON字符串 转换为 Audio  异常:{} ;str:{}", e.getMessage(), str);
+                throw new BaseException(format);
+            }
+    }
+    // 自定义反序列化器
+    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());
+        }
+    }
+}

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncBackgroupGroupImageItemToRedis.java

@@ -53,7 +53,7 @@ public class SyncBackgroupGroupImageItemToRedis {
     @Scheduled(fixedDelay = 100000)
     @SchedulerLock(name = "launch:SyncBackgroupGroupImageItemToRedis:syncBackgroupItemPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncBackgroupItemPlan(){
-        System.out.println("start");
+
         Mono<List<BackgroupImageItem>> listMono = template
                 .select(Query.query(Criteria.empty())
                         .columns(COLUMNARRAY),BackgroupImageItem.class)
@@ -65,18 +65,21 @@ public class SyncBackgroupGroupImageItemToRedis {
                   var groupUsersIdMysql = y.stream()
                           .map(BackgroupImageItem::groupId)
                           .map(String::valueOf).collect(Collectors.toList());
+                  if (!y.isEmpty()) {
+                      y.forEach(groupUser-> setRedisValue(groupUser, String.valueOf(x)).subscribe());
+                  }
                   reactiveRedisTemplate
                           .opsForSet()
                           .members(BackgroupImageItemRedisKey.BACKGROUP_IMAGE_ITEM_INFO_SET+x)
                           .switchIfEmpty(Mono.just(""))
                           .collectList().flatMap(listIds -> {
-                      var addList = groupUsersIdMysql.stream().filter(f->!listIds.contains(f)).collect(Collectors.toList());
-                      if (!addList.isEmpty()) {
-                          var collect = y.stream()
-                                  .filter(f -> addList.contains(String.valueOf(f.groupId())))
-                                  .collect(Collectors.toList());
-                          collect.forEach(groupUser-> setRedisValue(groupUser, String.valueOf(x)).subscribe());
-                      }
+//                      var addList = groupUsersIdMysql.stream().filter(f->!listIds.contains(f)).collect(Collectors.toList());
+//                      if (!addList.isEmpty()) {
+//                          var collect = y.stream()
+//                                  .filter(f -> addList.contains(String.valueOf(f.groupId())))
+//                                  .collect(Collectors.toList());
+//                          collect.forEach(groupUser-> setRedisValue(groupUser, String.valueOf(x)).subscribe());
+//                      }
                       var deleteList = listIds.stream().filter(f->!groupUsersIdMysql.contains(f)).collect(Collectors.toList());
                       if (!deleteList.isEmpty()) {
                           deleteList.forEach(id-> executeDelete(id, String.valueOf(x)).subscribe());
@@ -87,7 +90,7 @@ public class SyncBackgroupGroupImageItemToRedis {
               ));
               return Mono.empty();
         }).subscribe();
-        System.out.println("end");
+
     }
 
     @NotNull

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncGroupUsersToRedis.java

@@ -67,18 +67,21 @@ public class SyncGroupUsersToRedis {
                           var groupUsersIdMysql = y.stream()
                                   .map(GroupUsers::groupUsersId)
                                   .map(String::valueOf).collect(Collectors.toList());
+                          if (!y.isEmpty()) {
+                              y.forEach(groupUser->setRedisValue(groupUser, String.valueOf(x)).subscribe());
+                          }
                           reactiveRedisTemplate
                                   .opsForSet()
                                   .members(GroupUsersRedisKey.GROUP_USERS_INFO_SET+x)
                                   .switchIfEmpty(Mono.just(""))
                                   .collectList().flatMap(listIds -> {
-                              var addList = groupUsersIdMysql.stream().filter(f->!listIds.contains(f)).collect(Collectors.toList());
-                              if (!addList.isEmpty()) {
-                                  List<GroupUsers> collect = y.stream()
-                                          .filter(f -> addList.contains(String.valueOf(f.groupUsersId())))
-                                          .collect(Collectors.toList());
-                                  collect.forEach(groupUser-> setRedisValue(groupUser, String.valueOf(x)).subscribe());
-                              }
+//                              var addList = groupUsersIdMysql.stream().filter(f->!listIds.contains(f)).collect(Collectors.toList());
+//                              if (!addList.isEmpty()) {
+//                                  List<GroupUsers> collect = y.stream()
+//                                          .filter(f -> addList.contains(String.valueOf(f.groupUsersId())))
+//                                          .collect(Collectors.toList());
+//                                  collect.forEach(groupUser-> setRedisValue(groupUser, String.valueOf(x)).subscribe());
+//                              }
                               var deleteList = listIds.stream().filter(f->!groupUsersIdMysql.contains(f)).collect(Collectors.toList());
                               if (!deleteList.isEmpty()) {
                                   deleteList.forEach(groupUser-> executeDelete(groupUser, String.valueOf(x)).subscribe());

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncImageToRedis.java

@@ -58,7 +58,8 @@ public class SyncImageToRedis {
 //              .and("group_type_id").is(f.groupId())
 //              .and("deleted_at").isNull()), Image.class)
 //              .onErrorResume(throwable -> Mono.error(new BaseException("获取群封面异常"))).collectList();
-      @Scheduled(fixedDelay = 120000)
+
+    @Scheduled(fixedDelay = 120000)
     @SchedulerLock(name = "launch:SyncImageToRedis:syncImagePlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncImagePlan(){
         Query query = Query.query(Criteria
@@ -71,10 +72,12 @@ public class SyncImageToRedis {
                 .collectList();
         listMono.flatMap(list -> {
             //根据groupId分组 在根据type分组
-            Map<Integer, List<Image>> collect1 = list.stream().collect(Collectors.groupingBy(Image::groupTypeId));
-            collect1.forEach(((x,y)->{
-                Map<Integer, List<Image>> collect = y.stream().collect(Collectors.groupingBy(Image::type));
-                collect.forEach((x1,y1)-> y1.forEach(image-> setRedisValue(image, x+"-"+x1).subscribe())); }));
+            list.forEach(image-> setRedisValue(image, image.groupTypeId()+"-"+image.type()).subscribe());
+//            Map<Integer, List<Image>> collect1 = list.stream().collect(Collectors.groupingBy(Image::groupTypeId));
+//            collect1.forEach(((x,y)->{
+//                Map<Integer, List<Image>> collect = y.stream().collect(Collectors.groupingBy(Image::type));
+//                collect.forEach((x1,y1)-> y1.forEach(image-> setRedisValue(image, x+"-"+x1).subscribe()));
+//            }));
             return Mono.empty();
         }).subscribe();
         Query queryDelete = Query.query(Criteria
@@ -87,13 +90,15 @@ public class SyncImageToRedis {
                 .collectList();
         listMonoDe.flatMap(list -> {
             //根据groupId分组 在根据type分组
-            Map<Integer, List<Image>> collect1 = list.stream().collect(Collectors.groupingBy(Image::groupTypeId));
-            collect1.forEach(((x,y)->{
-                Map<Integer, List<Image>> collect = y.stream().collect(Collectors.groupingBy(Image::type));
-                collect.forEach((x1,y1)-> y1.forEach(image->
-                        deleteRedisValue(image, x+"-"+x1).subscribe()
-                ));
-            }));
+            list.forEach(image-> deleteRedisValue(image, image.groupTypeId()+"-"+image.type()).subscribe());
+
+//            Map<Integer, List<Image>> collect1 = list.stream().collect(Collectors.groupingBy(Image::groupTypeId));
+//            collect1.forEach(((x,y)->{
+//                Map<Integer, List<Image>> collect = y.stream().collect(Collectors.groupingBy(Image::type));
+//                collect.forEach((x1,y1)-> y1.forEach(image->
+//                        deleteRedisValue(image, x+"-"+x1).subscribe()
+//                ));
+//            }));
             return Mono.empty();
         }).subscribe();
     }

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncListeningAudioToRedis.java

@@ -71,12 +71,8 @@ public class SyncListeningAudioToRedis {
                 var rids = tupleN.getT2();
                 var planNews = tupleN.getT3();
                 // +
-                var addIds = mids.stream().filter(mid -> !rids.contains(mid)).collect(Collectors.toList());
-                if (!addIds.isEmpty()) {
-                    List<ListeningAudio> addPlanNew = planNews.stream().filter(f -> addIds.contains(getSetValue(f))).collect(Collectors.toList());
-                    if (!addPlanNew.isEmpty()) {
-                        Flux.fromIterable(addPlanNew).flatMap(this::setRedisValue).subscribe();
-                    }
+                if (!planNews.isEmpty()) {
+                    Flux.fromIterable(planNews).flatMap(this::setRedisValue).subscribe();
                 }
                 // -
                 var removeIds = rids.stream().filter(rid -> !mids.contains(rid)).collect(Collectors.toList());
@@ -92,7 +88,7 @@ public class SyncListeningAudioToRedis {
     public Mono<Void> setRedisValue(ListeningAudio planNew) {
         String stringBuilder = getSetValue(planNew);
         String json= getPlanNewToString(planNew);
-        reactiveRedisTemplate.opsForHash().put(LinteningAudioRedisKey.lintening_Audio_INFO_HASH , stringBuilder, json).subscribe();
+        reactiveRedisTemplate.opsForHash().put(LinteningAudioRedisKey.LINTENING_AUDIO_INFO_HASH , stringBuilder, json).subscribe();
         reactiveRedisTemplate.opsForSet().add(LinteningAudioRedisKey.PLAN_NEW_INFO_SET,stringBuilder ).subscribe();
        return Mono.empty();
     }
@@ -138,13 +134,13 @@ public class SyncListeningAudioToRedis {
     public Mono<Void> deleteRedis(ListeningAudio planNew) {
         String setValue = getSetValue(planNew);
         reactiveRedisTemplate.opsForSet().remove(LinteningAudioRedisKey.PLAN_NEW_INFO_SET, setValue).subscribe();
-        reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.lintening_Audio_INFO_HASH, setValue).subscribe();
+        reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.LINTENING_AUDIO_INFO_HASH, setValue).subscribe();
         return Mono.empty();
     }
     @NotNull
     public Mono<Void> getDelete(String key) {
          reactiveRedisTemplate.opsForSet().remove(LinteningAudioRedisKey.PLAN_NEW_INFO_SET, key).subscribe();
-         reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.lintening_Audio_INFO_HASH, key).subscribe();
+         reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.LINTENING_AUDIO_INFO_HASH, key).subscribe();
         return Mono.empty();
     }
 
@@ -187,7 +183,7 @@ public class SyncListeningAudioToRedis {
                               .collect(Collectors.toList());
                       return Flux.fromIterable(collect).flatMap(f -> reactiveRedisTemplate
                               .opsForHash()
-                              .get(LinteningAudioRedisKey.lintening_Audio_INFO_HASH , f))
+                              .get(LinteningAudioRedisKey.LINTENING_AUDIO_INFO_HASH , f))
                               .flatMap(str -> {
                                   if (Objects.nonNull(str) && !"".equals(str)) {
                                     return   Mono.just(strParsePlanNew(String.valueOf(str)));

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncPlanNewCommonListenToRedis.java

@@ -0,0 +1,199 @@
+package com.webflux.launchadmin.mysql.task;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.plumelog.core.util.LogExceptionStackTrace;
+import com.webflux.launchadmin.global.BaseException;
+import com.webflux.launchadmin.mysql.entity.planNew.PlanNewCommon;
+import com.weblux.launchredis.utils.PlanNewCommonRedisKey;
+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 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.Objects;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
+public class SyncPlanNewCommonListenToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+    private static final String COLUMN="id,name,code,type,poster_template_id,group_id,config_json,import_domain,preview_url";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+    @Scheduled(fixedDelay = 170000)
+    @SchedulerLock(name = "launch:SyncPlanNewCommonListenToRedis:syncPlanNewCommonPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
+    public void syncPlanNewCommonPlan(){
+        Mono<List<PlanNewCommon>> listMono = template
+                .select(Query.query(Criteria.empty())
+                        .columns(COLUMNARRAY),PlanNewCommon.class)
+                .collectList();
+          listMono.flatMap(list -> {
+            List<String> collect = list.stream().map(PlanNewCommon::code).collect(Collectors.toList());
+            Mono<List<String>> listMono1 = reactiveRedisTemplate
+                    .opsForSet()
+                    .members(PlanNewCommonRedisKey.PLAN_NEW_COMMON_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();
+                // + 全部添加一遍 预防构建宕机 数据不一致 省略数据set hash 一致性判断
+                if (!planNews.isEmpty()) { //增加io 减少逻辑 保证数据一致性
+                    Flux.fromIterable(planNews)
+                            .flatMap(this::setRedisValue).collectList().subscribe();
+                }
+//                List<String> addIds = mids.stream().filter(mid -> !rids.contains(mid)).collect(Collectors.toList());
+//                if (!addIds.isEmpty()) {
+//                    List<PlanNewCommon> addPlanNew = planNews.stream().filter(f -> addIds.contains(f.code())).collect(Collectors.toList());
+//                    if (!addPlanNew.isEmpty()) {
+//                      Flux.fromIterable(addPlanNew)
+//                                .flatMap(this::setRedisValue).collectList().subscribe();
+//                    }
+//                }
+                // -
+                List<String> removeIds = rids.stream().filter(rid -> !mids.contains(rid)).collect(Collectors.toList());
+                if (!removeIds.isEmpty()) {
+                    Flux.fromIterable(removeIds).flatMap(this::getDelete).subscribe();
+                }
+                return Mono.empty();
+            }).then();
+        }).subscribe();
+    }
+
+
+    @NotNull
+    public Mono<Void> setRedisValue(PlanNewCommon planNew) {
+        reactiveRedisTemplate.opsForSet().add(PlanNewCommonRedisKey.PLAN_NEW_COMMON_INFO_SET, planNew.code()).subscribe();
+        String json= getPlanNewToString(planNew);
+        reactiveRedisTemplate.opsForHash().put(PlanNewCommonRedisKey.PLAN_NEW_COMMON_INFO_HASH,planNew.code(), json).subscribe();
+       return Mono.empty();
+    }
+    @NotNull
+    public Mono<Void> getDelete(String code) {
+        reactiveRedisTemplate.opsForSet().remove(PlanNewCommonRedisKey.PLAN_NEW_COMMON_INFO_SET, code).subscribe();
+        reactiveRedisTemplate.opsForHash().remove(PlanNewCommonRedisKey.PLAN_NEW_COMMON_INFO_HASH , code).subscribe();
+        return Mono.empty();
+    }
+
+    private String getPlanNewToString(PlanNewCommon 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());
+        }
+    }
+
+
+
+
+
+    /**
+     * 通过code从数据库获取PlanNew并缓存redis
+     * @param code
+     * @return
+     */
+    public Mono<PlanNewCommon> byCodeAndMysqlGetPlanNew(String code){
+        Mono<PlanNewCommon> planNewMono = template
+                .selectOne(Query.query(Criteria.where("code").is(code))
+                        .columns(COLUMNARRAY), PlanNewCommon.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 -> {
+            setRedisValue(planNew).subscribe();
+           return Mono.just(planNew);
+        });
+    }
+
+    /**
+     * 通过redis获取缓存 空或失败 从mysql中取
+     * @param code
+     * @return
+     */
+    public Mono<PlanNewCommon> byRedisGetPlanNewCommonIfErByMysql(String code){
+      return   reactiveRedisTemplate.opsForHash()
+                 .get(PlanNewCommonRedisKey.PLAN_NEW_COMMON_INFO_HASH , code)
+                 .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis PlanNewCommon 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return Mono.just("");
+              })
+              .flatMap(s -> {
+                     Mono<PlanNewCommon> planNewMono;
+                     if(!"".equals(s) && Objects.nonNull(s)){
+                         try {
+                             PlanNewCommon planNew = strParsePlanNew(String.valueOf(s));
+                             planNewMono=Mono.just(planNew);
+                         }catch (Exception e){
+                             log.error("接口异常  {} e = {}","获取redis Value 解析PlanNewCommon失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e));
+                             planNewMono = byCodeAndMysqlGetPlanNew(code);
+                         }
+                     }else {
+                         planNewMono = byCodeAndMysqlGetPlanNew(code);
+                     }
+                     return planNewMono;
+         });
+    }
+    /**
+     * str 转PlanNew
+     * @param str
+     * @return
+     */
+    public PlanNewCommon 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, PlanNewCommon.class);
+            } catch (JsonProcessingException e) {
+                throw new BaseException(" 将 JSON字符串 转换为 PlanNewCommon  异常：：："+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());
+        }
+    }
+}

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncPlanNewListenToRedis.java

@@ -54,7 +54,7 @@ public class SyncPlanNewListenToRedis {
     @Scheduled(fixedDelay = 160000)
     @SchedulerLock(name = "launch:SyncPlanNewListenToRedis:syncLinteningPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncLinteningPlan(){
-        System.out.println("start");
+
         Mono<List<ListeningPlanNew>> listMono = template
                 .select(Query.query(Criteria.empty())
                         .columns(COLUMNARRAY),ListeningPlanNew.class)
@@ -86,7 +86,7 @@ public class SyncPlanNewListenToRedis {
                 return Mono.empty();
             }).then();
         }).subscribe();
-        System.out.println("end");
+
     }
 
 
@@ -116,10 +116,6 @@ public class SyncPlanNewListenToRedis {
         }
     }
 
-
-
-
-
     /**
      * 通过code从数据库获取PlanNew并缓存redis
      * @param code

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncPlanNewToRedis.java

@@ -31,6 +31,7 @@ import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 @Component
@@ -53,32 +54,37 @@ public class SyncPlanNewToRedis {
             "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";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
     /**
      *同步裂变计划
      */
     @Scheduled(fixedDelay = 170000)
     @SchedulerLock(name = "launch:SyncPlanNewToRedis:syncFissionPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncFissionPlan(){
-        Mono<List<PlanNew>> listMono = template.select(PlanNew.class).all().collectList();
+        Mono<List<PlanNew>> listMono = template.select(Query.query(Criteria.empty()).columns(COLUMNARRAY),PlanNew.class).collectList();
         listMono.flatMap(list -> {
+            if (!list.isEmpty()) {
+                Flux.fromIterable(list)
+                        .flatMap(this::setRedisValue).then().subscribe();
+            }
             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->{
+            return Mono.zip(Mono.just(collect),listMono1).flatMap(tupleN->{
                 var mids = tupleN.getT1();
                 var rids = tupleN.getT2();
-                var planNews = tupleN.getT3();
+             //   var planNews = tupleN.getT3();
                 // +
-                List<String> addIds = mids.stream().filter(mid -> !rids.contains(mid)).collect(Collectors.toList());
-                if (!addIds.isEmpty()) {
-                    List<PlanNew> addPlanNew = planNews.stream().filter(f -> addIds.contains(f.code())).collect(Collectors.toList());
-                    if (!addPlanNew.isEmpty()) {
-                        Flux.fromIterable(addPlanNew)
-                                .flatMap(this::setRedisValue).then().subscribe();
-                    }
-                }
+//                List<String> addIds = mids.stream().filter(mid -> !rids.contains(mid)).collect(Collectors.toList());
+//                if (!addIds.isEmpty()) {
+//                    List<PlanNew> addPlanNew = planNews.stream().filter(f -> addIds.contains(f.code())).collect(Collectors.toList());
+//                    if (!addPlanNew.isEmpty()) {
+//                        Flux.fromIterable(addPlanNew)
+//                                .flatMap(this::setRedisValue).then().subscribe();
+//                    }
+//                }
                 // -
                 List<String> removeIds = rids.stream().filter(rid -> !mids.contains(rid)).collect(Collectors.toList());
                 if (!removeIds.isEmpty()) {
@@ -94,8 +100,8 @@ public class SyncPlanNewToRedis {
         reactiveRedisTemplate.opsForSet().add(FissionRedisKey.PLAN_NEW_INFO_SET, planNew.code()).subscribe();
         String json= getPlanNewToString(planNew);
         reactiveRedisTemplate
-                .opsForValue()
-                .set(FissionRedisKey.PLAN_NEW_INFO_KEY + planNew.code(), json).subscribe();
+                .opsForHash()
+                .put(FissionRedisKey.PLAN_NEW_INFO_HASH , planNew.code(), json).subscribe();
        return Mono.empty();
     }
 
@@ -116,7 +122,7 @@ public class SyncPlanNewToRedis {
     @NotNull
     public Mono<Void> getDelete(String code) {
        reactiveRedisTemplate.opsForSet().remove(FissionRedisKey.PLAN_NEW_INFO_SET, code).subscribe();
-       reactiveRedisTemplate.opsForValue().delete(FissionRedisKey.PLAN_NEW_INFO_KEY + code).subscribe();
+       reactiveRedisTemplate.opsForHash().remove(FissionRedisKey.PLAN_NEW_INFO_HASH , code).subscribe();
        return Mono.empty();
     }
 
@@ -142,15 +148,15 @@ public class SyncPlanNewToRedis {
      * @return
      */
     public Mono<PlanNew> byRedisGetPlanNewIfErByMysql(String code){
-      return   reactiveRedisTemplate.opsForValue()
-                 .get(FissionRedisKey.PLAN_NEW_INFO_KEY + code)
+      return   reactiveRedisTemplate.opsForHash()
+                 .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)){
+                     if(Objects.nonNull(s) && !"".equals(s)){
                          try {
-                             PlanNew planNew = strParsePlanNew(s);
+                             PlanNew planNew = strParsePlanNew(String.valueOf(s));
                              planNewMono=Mono.just(planNew);
                          }catch (Exception e){
                              log.error("接口异常  {} e = {}","获取redis Value 解析PlanNew失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e));

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncPosterTempAudioToRedis.java

@@ -0,0 +1,199 @@
+package com.webflux.launchadmin.mysql.task;
+
+import cn.hutool.core.util.StrUtil;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.plumelog.core.util.LogExceptionStackTrace;
+import com.webflux.launchadmin.global.BaseException;
+import com.webflux.launchadmin.mysql.entity.planNew.PosterTemplate;
+import com.weblux.launchredis.utils.PosterTempRedisKey;
+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.domain.Sort;
+import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.core.ScanOptions;
+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 java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
+public class SyncPosterTempAudioToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+
+    private static final String COLUMN="poster_template_id,poster_template_name";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+    @Scheduled(fixedDelay = 130000)
+    @SchedulerLock(name = "launch:SyncPosterTempAudioToRedis:syncPosterTemplatePlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
+    public void syncPosterTemplatePlan(){
+        Mono<List<PosterTemplate>> listMono = template.select(Query.query(Criteria.empty())
+                        .columns(COLUMNARRAY)
+                , PosterTemplate.class).collectList();
+        listMono.flatMap(list -> {
+            var collect = list.stream().map(PosterTemplate::posterTemplateId).map(String::valueOf).collect(Collectors.toList());
+            Mono<List<String>> listMono1 = reactiveRedisTemplate
+                    .opsForSet()
+                    .members(PosterTempRedisKey.POSTER_TEMP_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();
+                // +
+                if (!planNews.isEmpty()) {
+                    Flux.fromIterable(planNews).flatMap(this::setRedisValue).subscribe();
+                }
+                // -
+                var removeIds = rids.stream().filter(rid -> !mids.contains(rid)).collect(Collectors.toList());
+                if (!removeIds.isEmpty()) {
+                    Flux.fromIterable(removeIds).flatMap(this::getDelete).subscribe();
+                }
+              return  Mono.empty();
+            });
+        }).subscribe();
+    }
+
+    @NotNull
+    public Mono<Void> setRedisValue(PosterTemplate planNew) {
+        String json= getPlanNewToString(planNew);
+        reactiveRedisTemplate.opsForHash().put(PosterTempRedisKey.POSTER_TEMP_INFO_HASH , String.valueOf(planNew.posterTemplateId()), json).subscribe();
+        reactiveRedisTemplate.opsForSet().add(PosterTempRedisKey.POSTER_TEMP_INFO_SET,String.valueOf(planNew.posterTemplateId()) ).subscribe();
+       return Mono.empty();
+    }
+
+
+
+
+    private String getPlanNewToString(PosterTemplate 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());
+        }
+    }
+
+
+    public Mono<Void> deleteRedis(PosterTemplate planNew) {
+        reactiveRedisTemplate.opsForSet().remove(PosterTempRedisKey.POSTER_TEMP_INFO_SET, String.valueOf(planNew.posterTemplateId())).subscribe();
+        reactiveRedisTemplate.opsForHash().remove(PosterTempRedisKey.POSTER_TEMP_INFO_HASH, String.valueOf(planNew.posterTemplateId())).subscribe();
+        return Mono.empty();
+    }
+    @NotNull
+    public Mono<Void> getDelete(String key) {
+         reactiveRedisTemplate.opsForSet().remove(PosterTempRedisKey.POSTER_TEMP_INFO_SET, key).subscribe();
+         reactiveRedisTemplate.opsForHash().remove(PosterTempRedisKey.POSTER_TEMP_INFO_HASH, key).subscribe();
+        return Mono.empty();
+    }
+
+    /**
+     * 通过pageShow从数据库获取 并缓存redis
+     * @param id
+     * @return
+     */
+    public Mono<PosterTemplate> byCodeAndMysqlGetPosterTemplate(Integer id){
+        Mono<PosterTemplate> idx_num = template.selectOne(Query.query(Criteria.where("poster_template_id").is(id))
+                .columns(COLUMNARRAY)
+               , PosterTemplate.class)
+                .switchIfEmpty(Mono.error(new BaseException(StrUtil.format("通过海报模版id：{} 查询结果为空",id))))
+                .onErrorResume(throwable -> Mono.error(new BaseException("获取海报模版err"+throwable.getMessage())));
+       return idx_num.flatMap(posterTemplate -> {
+           setRedisValue(posterTemplate).subscribe();
+           return Mono.just(posterTemplate);
+        });
+    }
+
+    /**
+     * 通过redis获取缓存 空或失败 从mysql中取
+     * @param
+     * @return
+     */
+    public Mono<PosterTemplate> byRedisGetLinteningAudioIfErByMysql(Integer id){
+      return   reactiveRedisTemplate
+              .opsForHash()
+              .get(PosterTempRedisKey.POSTER_TEMP_INFO_HASH,id)
+              .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis PosterTemplate 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return Mono.just("");
+              })
+              .flatMap(s -> {
+                  Mono<PosterTemplate> planNewMono;
+                  if( Objects.nonNull(s) && !"".equals(s)){
+                      try {
+                          PosterTemplate planNew = strParsePlanNew(String.valueOf(s));
+                          planNewMono=Mono.just(planNew);
+                      }catch (Exception e){
+                          log.error("接口异常  {} e = {}","获取redis Value 解析PosterTemplate失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e));
+                          planNewMono = byCodeAndMysqlGetPosterTemplate(id);
+                      }
+                  }else {
+                      planNewMono = byCodeAndMysqlGetPosterTemplate(id);
+                  }
+                  return planNewMono;
+         });
+    }
+    /**
+     * str 转PlanNew
+     * @param str
+     * @return
+     */
+    public PosterTemplate 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, PosterTemplate.class);
+            } catch (JsonProcessingException e) {
+                throw new BaseException(" 将 JSON字符串 转换为 PosterTemplate  异常：：："+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());
+        }
+    }
+}

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/SyncReturnLoopToRedis.java

@@ -0,0 +1,185 @@
+package com.webflux.launchadmin.mysql.task;
+
+import cn.hutool.core.util.StrUtil;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.plumelog.core.util.LogExceptionStackTrace;
+import com.webflux.launchadmin.global.BaseException;
+import com.webflux.launchadmin.mysql.entity.planNew.ReturnLoop;
+import com.weblux.launchredis.utils.ReturnLoopRedisKey;
+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 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.Objects;
+
+@Component
+@Slf4j
+public class SyncReturnLoopToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+
+    private static final String COLUMN="id,name,group_type,group_type_name,group_type_id,domain,is_show_list,return_list,return_number,type,domain_id";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+//    template.selectOne(Query.query(Criteria.where("id").is(id)), ReturnLoop.class)
+//            .switchIfEmpty(Mono.error(new BaseException(StrUtil.format("通过计划retReturnLoopId：{}查询音频为空", id))))
+//            .onErrorResume(throwable -> Mono.error(new BaseException(StrUtil.format("通过计划retReturnLoopId：{}查询音频Error::{}", id, throwable.getMessage()))));
+
+    @Scheduled(fixedDelay = 190000)
+    @SchedulerLock(name = "launch:SyncReturnLoopToRedis:syncReturnLoopPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
+    public void syncReturnLoopPlan(){
+        Mono<List<ReturnLoop>> listMono = template
+                .select(Query.query(Criteria.where("deleted_at").isNull())
+                        .columns(COLUMNARRAY)
+                , ReturnLoop.class).collectList();
+        listMono.flatMap(list -> {
+            if (!list.isEmpty()) {
+                Flux.fromIterable(list).flatMap(this::setRedisValue).subscribe();
+            }
+            return Mono.empty();
+        }).subscribe();
+        Mono<List<ReturnLoop>> listMonoDelette = template
+                .select(Query.query(Criteria.where("deleted_at").isNotNull())
+                                .columns(COLUMNARRAY)
+                        , ReturnLoop.class).collectList();
+        listMonoDelette.flatMap(list -> {
+            if (!list.isEmpty()) {
+                Flux.fromIterable(list).flatMap(this::deleteRedis).subscribe();
+            }
+            return Mono.empty();
+        }).subscribe();
+    }
+
+    @NotNull
+    public Mono<Void> setRedisValue(ReturnLoop audio) {
+        String json= getPlanNewToString(audio);
+        reactiveRedisTemplate.opsForHash().put(ReturnLoopRedisKey.RETURN_LOOP_INFO_HASH , String.valueOf(audio.id()), json).subscribe();
+       return Mono.empty();
+    }
+
+    private String getPlanNewToString(ReturnLoop 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());
+        }
+    }
+
+    public Mono<Void> deleteRedis(ReturnLoop audio) {
+        reactiveRedisTemplate.opsForHash().remove(ReturnLoopRedisKey.RETURN_LOOP_INFO_HASH, String.valueOf(audio.id())).subscribe();
+        return Mono.empty();
+    }
+
+    /**
+     * 通过pageShow从数据库获取 并缓存redis
+     * @param id
+     * @return
+     */
+    public Mono<ReturnLoop> byIdFromMysqlGetReturnLoop(Long id){
+        Mono<ReturnLoop> id1 = template.selectOne(Query.query(Criteria.where("id").is(id)
+                .and("deleted_at").isNull()), ReturnLoop.class)
+                .switchIfEmpty(Mono.just(getReturnLoop()))
+                .onErrorResume(throwable ->
+                        Mono.error(new BaseException(StrUtil.format("通过计划retReturnLoopId：{}查询Error::{}", id, throwable.getMessage()))));
+        return id1.flatMap(audio -> {
+            setRedisValue(audio).subscribe();
+           return Mono.just(audio);
+        });
+    }
+    private ReturnLoop getReturnLoop() {
+        return new ReturnLoop(-1L, null, null, null, null, null, null, null, null, null, null, null, null, null);
+    }
+    /**
+     * 通过redis获取缓存 空或失败 从mysql中取
+     * @param
+     * @return
+     */
+    public Mono<ReturnLoop> byRedisGetReturnLoopIfErByMysql(Long id){
+      return   reactiveRedisTemplate
+              .opsForHash()
+              .get(ReturnLoopRedisKey.RETURN_LOOP_INFO_HASH,id)
+              .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis ReturnLoop 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return Mono.just("");
+              })
+              .flatMap(s -> {
+                  Mono<ReturnLoop> planNewMono;
+                  if(!"".equals(s) && Objects.nonNull(s)){
+                      try {
+                          ReturnLoop planNew = strParsePlanNew(String.valueOf(s));
+                          planNewMono=Mono.just(planNew);
+                      }catch (Exception e){
+                          log.error("接口异常  {} e = {} id={}","获取redis Value 解析ReturnLoop失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e),id);
+                          planNewMono = byIdFromMysqlGetReturnLoop(id);
+                      }
+                  }else {
+                      planNewMono = byIdFromMysqlGetReturnLoop(id);
+                  }
+                  return planNewMono;
+         });
+    }
+    /**
+     * str 解析对象
+     * @param str
+     * @return
+     */
+    public ReturnLoop 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, ReturnLoop.class);
+            } catch (JsonProcessingException e) {
+                String format = StrUtil.format(" 将 JSON字符串 转换为 ReturnLoop  异常:{} ;str:{}", e.getMessage(), str);
+                throw new BaseException(format);
+            }
+    }
+    // 自定义反序列化器
+    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());
+        }
+    }
+}

launch-admin/src/main/java/com/webflux/launchadmin/mysql/task/config/SchedulingConfig.java

@@ -0,0 +1,31 @@
+package com.webflux.launchadmin.mysql.task.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Executor;
+
+@Configuration
+@EnableScheduling
+public class SchedulingConfig {
+
+    @Bean
+    public Executor virtualThreadExecutor() {
+        return Executors.newVirtualThreadPerTaskExecutor();
+    }
+
+    @Bean
+    public ThreadPoolTaskScheduler taskScheduler(Executor virtualThreadExecutor) {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler() {
+            @Override
+            public void execute(Runnable task) {
+                virtualThreadExecutor.execute(task);
+            }
+        };
+        scheduler.setPoolSize(2);//不受限制
+        scheduler.setThreadNamePrefix("scheduled-task-");
+        return scheduler;
+    }
+}

launch-redis/src/main/java/com/weblux/launchredis/utils/AudioRedisKey.java

@@ -0,0 +1,17 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+public class AudioRedisKey {
+
+
+    //裂变计划缓存
+    public final static String AUDIO_INFO = "sync_audio_info";
+    public final static String AUDIO_INFO_SET = "sync_audio_info_set_ids";
+    public final static String AUDIO_INFO_KEY = "sync_audio_info_";
+    public final static String AUDIO_INFO_HASH = "sync_Audio_info_hash_json";
+}

launch-redis/src/main/java/com/weblux/launchredis/utils/FissionRedisKey.java

@@ -13,4 +13,5 @@ public class FissionRedisKey {
     public final static String PLAN_NEW_INFO = "sync_plan_new_info";
     public final static String PLAN_NEW_INFO_SET = "sync_plan_new_info_set_ids";
     public final static String PLAN_NEW_INFO_KEY = "sync_plan_new_info_";
+    public final static String PLAN_NEW_INFO_HASH = "sync_plan_new_info_hash_json";
 }

launch-redis/src/main/java/com/weblux/launchredis/utils/LinteningAudioRedisKey.java

@@ -13,5 +13,5 @@ public class LinteningAudioRedisKey {
     public final static String PLAN_NEW_INFO = "sync_lintening_audio_info";
     public final static String PLAN_NEW_INFO_SET = "sync_lintening_audio_info_set_ids";
     public final static String PLAN_NEW_INFO_KEY = "sync_lintening_audio_info_";
-    public final static String lintening_Audio_INFO_HASH = "sync_lintening_Audio_info_hash_json";
+    public final static String LINTENING_AUDIO_INFO_HASH = "sync_lintening_Audio_info_hash_json";
 }

launch-redis/src/main/java/com/weblux/launchredis/utils/PlanNewCommonRedisKey.java

@@ -0,0 +1,17 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+public class PlanNewCommonRedisKey {
+
+
+    //裂变计划缓存
+    public final static String PLAN_NEW_COMMON_INFO = "sync_plan_new_common_info";
+    public final static String PLAN_NEW_COMMON_INFO_SET = "sync_plan_new_common_info_set_ids";
+    public final static String PLAN_NEW_COMMON_INFO_KEY = "sync_plan_new_common_info_";
+    public final static String PLAN_NEW_COMMON_INFO_HASH = "sync_plan_new_common_info_hash_json";
+}

launch-redis/src/main/java/com/weblux/launchredis/utils/PosterTempRedisKey.java

@@ -0,0 +1,17 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+public class PosterTempRedisKey {
+
+
+    //裂变计划缓存
+    public final static String POSTER_TEMP_INFO = "sync_poster_temp_info";
+    public final static String POSTER_TEMP_INFO_SET = "sync_poster_temp_info_set_ids";
+    public final static String POSTER_TEMP_INFO_KEY = "sync_poster_temp_info_";
+    public final static String POSTER_TEMP_INFO_HASH = "sync_poster_temp_info_hash_json";
+}

launch-redis/src/main/java/com/weblux/launchredis/utils/ReturnLoopRedisKey.java

@@ -0,0 +1,17 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+public class ReturnLoopRedisKey {
+
+
+    //裂变计划缓存
+    public final static String RETURN_LOOP_INFO = "sync_return_loop_info";
+    public final static String RETURN_LOOP_INFO_SET = "sync_return_loop_info_set_ids";
+    public final static String RETURN_LOOP_INFO_KEY = "sync_return_loop_info_";
+    public final static String RETURN_LOOP_INFO_HASH = "sync_return_loop_info_hash_json";
+}