裂变计划redis v0.02 +听书

(cherry picked from commit f4ee91f4ff94a60f023d4fd6d4b47774035584d5)

launch-admin/src/main/java/com/webflux/launchadmin/mysql/controller/testController.java

@@ -6,8 +6,12 @@ import cn.hutool.json.JSONUtil;
 
 import com.webflux.launchadmin.context.UserContext;
 import com.webflux.launchadmin.context.XxlSsoUser;
+import com.webflux.launchadmin.mysql.entity.listening.ListeningAudio;
+import com.webflux.launchadmin.mysql.entity.listening.ListeningPlanNew;
 import com.webflux.launchadmin.mysql.entity.test.Tweet;
 import com.webflux.launchadmin.mysql.repository.test.TweetRepository;
+import com.webflux.launchadmin.mysql.task.SyncListeningAudioToRedis;
+import com.webflux.launchadmin.mysql.task.SyncPlanNewListenToRedis;
 import com.webflux.launchcommon.returnObj.Paged;
 import com.webflux.launchcommon.returnObj.RData;
 import com.webflux.launchcommon.returnObj.RStatus;
@@ -43,7 +47,7 @@ import java.util.Objects;
  * 测试
  */
 @RestController
-@RequestMapping("test")
+@RequestMapping("out/test")
 @Slf4j
 public class testController {
 
@@ -53,29 +57,17 @@ public class testController {
     private TweetRepository tweetRepository;
     @Resource
     private R2dbcEntityTemplate template;
+    @Resource
+    private SyncListeningAudioToRedis redis;
 
     @GetMapping("get5")
-    public void  get5()   {
-        // Notice: the query string must be URL encoded 172.18.71.27:3306/launch_platform
-//        ConnectionFactory connectionFactory = ConnectionFactories.get("r2dbcs:mysql://launch_platform:launch_platform@172.18.71.27:3306/launch_platform");
-//        R2dbcEntityTemplate template = new R2dbcEntityTemplate(connectionFactory);
-//        Mono<Tweet> joe = template.insert(new Tweet(null, "Joe"));
-//        joe.subscribe(System.out::println);
-
-
+    public Mono<List<ListeningAudio>>  get5()   {
+        return redis.byRedisGetLinteningAudioIfErByMysql(3);
     }
     @GetMapping("get4")
-    public void  get4()   {
-        // url: r2dbcs:mysql://172.18.71.27:3306/launch_platform?serverZoneId=GMT%2b8&useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8  # r2dbcs:mysql://127.0.0.1:3306/test?serverZoneId=GMT%2b8&characterEncoding=UTF-8
-        //    username: launch_platform # root
-        //    password: launch_platform # 'QWERqwer!@#123'
-        // Notice: the query string must be URL encoded 172.18.71.27:3306/launch_platform
-//        ConnectionFactory connectionFactory = ConnectionFactories.get("r2dbcs:mysql://wx_share_liuwei:Wx_ShARe56578AwCH@rm-uf6xylut5n33hb3o5ro.mysql.rds.aliyuncs.com:3306/wx_share_test");
-//        R2dbcEntityTemplate template = new R2dbcEntityTemplate(connectionFactory);
-//        Mono<Tweet> joe = template.insert(new Tweet(10L, "Joe"));
-//        joe.subscribe(System.out::println);
-
-
+    public Mono<Void>  get4()   {
+        //return syncPlanNewListenToRedis.syncLinteningPlan();
+        return Mono.empty();
     }
 
 

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

@@ -0,0 +1,210 @@
+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.listening.ListeningAudio;
+import com.webflux.launchadmin.mysql.entity.listening.ListeningPlanNew;
+import com.weblux.launchredis.utils.LinteningAudioRedisKey;
+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.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 SyncListeningAudioToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+
+    private static final String COLUMN="new_audio_id,audio_name," +
+            "audio_image_url,audio_url,popup_node,idx_num";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+    @Scheduled(cron="0 */1 * * * ?")
+    @SchedulerLock(name = "launch:SyncListeningAudioToRedis:syncLinteningAudioPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
+    public void syncLinteningAudioPlan(){
+        Mono<List<ListeningAudio>> listMono = template.select(Query.query(Criteria.empty())
+                .sort(Sort.by(Sort.Order.asc("idx_num")))
+                        .columns(COLUMNARRAY)
+                , ListeningAudio.class).collectList();
+        listMono.flatMap(list -> {
+            var collect = list.stream().map(this::getSetValue).collect(Collectors.toList());
+            Mono<List<String>> listMono1 = reactiveRedisTemplate
+                    .opsForSet()
+                    .members(LinteningAudioRedisKey.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();
+                // +
+                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();
+                    }
+                }
+                // -
+                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(ListeningAudio planNew) {
+        String stringBuilder = getSetValue(planNew);
+        String json= getPlanNewToString(planNew);
+        reactiveRedisTemplate
+                .opsForValue()
+                .set(LinteningAudioRedisKey.PLAN_NEW_INFO_KEY + stringBuilder, json).subscribe();
+        reactiveRedisTemplate.opsForSet().add(LinteningAudioRedisKey.PLAN_NEW_INFO_SET,stringBuilder ).subscribe();
+       return Mono.empty();
+    }
+
+    @NotNull
+    private String getSetValue(ListeningAudio planNew) {
+        return planNew.idxNum() +
+                "-" +
+                planNew.newAudioId();
+    }
+
+    private String getPlanNewToString(ListeningAudio 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(ListeningAudio planNew) {
+        String setValue = getSetValue(planNew);
+        reactiveRedisTemplate.opsForSet().remove(LinteningAudioRedisKey.PLAN_NEW_INFO_SET, setValue).subscribe();
+        reactiveRedisTemplate.opsForValue().delete(LinteningAudioRedisKey.PLAN_NEW_INFO_KEY + setValue).subscribe();
+        return Mono.empty();
+    }
+    @NotNull
+    public Mono<Void> getDelete(String key) {
+        Mono<Long> remove = reactiveRedisTemplate.opsForSet().remove(LinteningAudioRedisKey.PLAN_NEW_INFO_SET, key);
+        Mono<Boolean> delete = reactiveRedisTemplate.opsForValue().delete(LinteningAudioRedisKey.PLAN_NEW_INFO_KEY + key);
+        return remove.then(delete).then();
+    }
+
+    /**
+     * 通过pageShow从数据库获取 并缓存redis
+     * @param pageShow
+     * @return
+     */
+    public Mono<List<ListeningAudio>> byCodeAndMysqlGetPlanNew(Integer pageShow){
+        Mono<List<ListeningAudio>> idx_num = template.select(Query.query(Criteria.empty())
+                .sort(Sort.by(Sort.Order.asc("idx_num")))
+                .columns(COLUMNARRAY)
+                .limit(pageShow), ListeningAudio.class).collectList();
+       return idx_num.flatMap(planNew -> {
+           Flux.fromIterable(planNew).flatMap(this::setRedisValue).subscribe();
+           return Mono.just(planNew);
+        });
+    }
+
+    /**
+     * 通过redis获取缓存 空或失败 从mysql中取
+     * @param
+     * @return
+     */
+    public Mono<List<ListeningAudio>> byRedisGetLinteningAudioIfErByMysql(Integer size){
+      return   reactiveRedisTemplate
+              .opsForSet()
+              .members(LinteningAudioRedisKey.PLAN_NEW_INFO_SET).collectList()
+              .switchIfEmpty(Mono.just(List.of()))
+              .flatMap(s -> {
+                  if (Objects.nonNull(s) && !s.isEmpty()) {
+                      var collect = s.stream().filter(f->Objects.nonNull(f)&&f.contains("-"))
+                              .sorted(Comparator.comparing((f)-> Integer.valueOf(f.split("-")[0]))).limit(size).collect(Collectors.toList());
+                      return Flux.fromIterable(collect).flatMap(f -> reactiveRedisTemplate
+                              .opsForValue()
+                              .get(LinteningAudioRedisKey.PLAN_NEW_INFO_KEY + f))
+                              .flatMap(str -> {
+                                  if (Objects.nonNull(str) && !"".equals(str)) {
+                                    return   Mono.just(strParsePlanNew(str));
+                                  }
+                                  return Mono.empty();
+
+                              }).collectList();
+                  }else {
+                     return byCodeAndMysqlGetPlanNew(size);
+                  }
+         });
+    }
+    /**
+     * str 转PlanNew
+     * @param str
+     * @return
+     */
+    public ListeningAudio 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, ListeningAudio.class);
+            } catch (JsonProcessingException e) {
+                throw new BaseException(" 将 JSON字符串 转换为 ListeningAudio  异常：：："+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

@@ -0,0 +1,202 @@
+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.listening.ListeningPlanNew;
+import com.webflux.launchadmin.mysql.entity.planNew.PlanNew;
+import com.weblux.launchredis.utils.LinteningRedisKey;
+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.Disposable;
+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.stream.Collectors;
+
+@Component
+@Slf4j
+public class SyncPlanNewListenToRedis {
+    @Resource
+    private R2dbcEntityTemplate template;
+    @Resource
+    private ReactiveRedisTemplate<String,String> reactiveRedisTemplate;
+
+    private static final String COLUMN="listening_plan_new_id,name,type,code," +
+            "group_id,group_name,poster_template_id,page_show,page_line,link," +
+            "background_group_id,baidu_dot,preview_url,list_button";
+    private static final String[] COLUMNARRAY=COLUMN.split(",");
+    /**
+     *同步裂变计划
+     */
+    @Scheduled(cron="0 */1 * * * ?")
+    @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)
+                .collectList();
+          listMono.flatMap(list -> {
+            List<String> collect = list.stream().map(ListeningPlanNew::code).collect(Collectors.toList());
+            Mono<List<String>> listMono1 = reactiveRedisTemplate
+                    .opsForSet()
+                    .members(LinteningRedisKey.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());
+                if (!addIds.isEmpty()) {
+                    List<ListeningPlanNew> 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();
+        System.out.println("end");
+    }
+    public Mono<List<String>> getIds(){
+        return reactiveRedisTemplate
+                .opsForSet()
+                .members(LinteningRedisKey.PLAN_NEW_INFO_SET).collectList()
+                .switchIfEmpty(Mono.just(List.of()));
+    }
+
+    @NotNull
+    public Mono<Void> setRedisValue(ListeningPlanNew planNew) {
+        reactiveRedisTemplate.opsForSet().add(LinteningRedisKey.PLAN_NEW_INFO_SET, planNew.code()).subscribe();
+        String json= getPlanNewToString(planNew);
+        reactiveRedisTemplate
+                .opsForValue()
+                .set(LinteningRedisKey.PLAN_NEW_INFO_KEY + planNew.code(), json).subscribe();
+       return Mono.empty();
+    }
+
+    private String getPlanNewToString(ListeningPlanNew 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) {
+         reactiveRedisTemplate.opsForSet().remove(LinteningRedisKey.PLAN_NEW_INFO_SET, code).subscribe();
+         reactiveRedisTemplate.opsForValue().delete(LinteningRedisKey.PLAN_NEW_INFO_KEY + code).subscribe();
+        return Mono.empty();
+    }
+
+    /**
+     * 通过code从数据库获取PlanNew并缓存redis
+     * @param code
+     * @return
+     */
+    public Mono<ListeningPlanNew> byCodeAndMysqlGetPlanNew(String code){
+        Mono<ListeningPlanNew> planNewMono = template
+                .selectOne(Query.query(Criteria.where("code").is(code))
+                        .columns(COLUMNARRAY), ListeningPlanNew.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<ListeningPlanNew> byRedisGetPlanNewIfErByMysql(String code){
+      return   reactiveRedisTemplate.opsForValue()
+                 .get(LinteningRedisKey.PLAN_NEW_INFO_KEY + code)
+                 .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())))
+              .flatMap(s -> {
+                     Mono<ListeningPlanNew> planNewMono;
+                     if(!"".equals(s)){
+                         try {
+                             ListeningPlanNew 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 ListeningPlanNew 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, ListeningPlanNew.class);
+            } catch (JsonProcessingException e) {
+                throw new BaseException(" 将 JSON字符串 转换为 ListeningPlanNew  异常：：："+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/SyncPlanNewToRedis.java

@@ -46,17 +46,18 @@ public class SyncPlanNewToRedis {
      * 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";
+      private static final 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")
+    //@Scheduled(cron="0 */1 * * * ?")
+   // @SchedulerLock(name = "launch:SyncPlanNewToRedis:syncFissionPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncFissionPlan(){
         Mono<List<PlanNew>> listMono = template.select(PlanNew.class).all().collectList();
         listMono.flatMap(list -> {
@@ -71,33 +72,31 @@ public class SyncPlanNewToRedis {
                 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();
+                        Flux.fromIterable(addPlanNew)
+                                .flatMap(this::setRedisValue).then().subscribe();
                     }
                 }
                 // -
-                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();
+                     Flux.fromIterable(removeIds).flatMap(this::getDelete).subscribe();
                 }
-              return  Mono.zip(addMono,then1).then();
+              return  Mono.empty();
             }).then();
-        });
+        }).subscribe();
     }
 
     @NotNull
     public Mono<Void> setRedisValue(PlanNew planNew) {
-        Mono<Long> add = reactiveRedisTemplate.opsForSet().add(FissionRedisKey.PLAN_NEW_INFO_SET, planNew.code());
+        reactiveRedisTemplate.opsForSet().add(FissionRedisKey.PLAN_NEW_INFO_SET, planNew.code()).subscribe();
         String json= getPlanNewToString(planNew);
-        Mono<Boolean> set = reactiveRedisTemplate
+        reactiveRedisTemplate
                 .opsForValue()
-                .set(FissionRedisKey.PLAN_NEW_INFO_KEY + planNew.code(), json);
-       return set.then(add).then();
+                .set(FissionRedisKey.PLAN_NEW_INFO_KEY + planNew.code(), json).subscribe();
+       return Mono.empty();
     }
 
     private String getPlanNewToString(PlanNew planNew) {
@@ -116,9 +115,9 @@ public class SyncPlanNewToRedis {
 
     @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();
+       reactiveRedisTemplate.opsForSet().remove(FissionRedisKey.PLAN_NEW_INFO_SET, code).subscribe();
+       reactiveRedisTemplate.opsForValue().delete(FissionRedisKey.PLAN_NEW_INFO_KEY + code).subscribe();
+       return Mono.empty();
     }
 
     /**

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

@@ -0,0 +1,16 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+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_";
+}

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

@@ -0,0 +1,16 @@
+package com.weblux.launchredis.utils;
+
+/**
+ * 裂变redis-key
+ *
+ * @author:
+ * @date: 2024/7/12 10:06
+ */
+public class LinteningRedisKey {
+
+
+    //裂变计划缓存
+    public final static String PLAN_NEW_INFO = "sync_lintening_plan_new_info";
+    public final static String PLAN_NEW_INFO_SET = "sync_lintening_plan_new_info_set_ids";
+    public final static String PLAN_NEW_INFO_KEY = "sync_lintening_plan_new_info_";
+}