Merge branch 'pro' of http://git.mokasz.com/liuwei/launch-webFlux-platform into pro

launch-admin/pom.xml

@@ -60,7 +60,11 @@
             <artifactId>hutool-all</artifactId>
             <version>5.3.7</version>
         </dependency>
-
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>2.12.3</version>
+        </dependency>
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>

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

@@ -132,8 +132,6 @@ public class ListeningPlanNewController {
      */
     @DeleteMapping("deleteAudioById/{newAudioId}")
     public Mono<RStatus<Long>> deleteAudioById (@PathVariable("newAudioId")  @NotNull Long newAudioId){
-        Mono<Long> listening_plan_new_id = template.delete(Query.query(Criteria.where("new_audio_id")
-                .is(newAudioId)), ListeningAudio.class);
-        return RStatus.success(listening_plan_new_id);
+        return  listeningPlanService.deleteAudioById(newAudioId);
     }
 }

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

@@ -33,11 +33,14 @@ import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.relational.core.query.Criteria;
 import org.springframework.data.relational.core.query.Query;
+
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.server.ServerWebExchange;
 import reactor.core.publisher.Mono;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
@@ -68,13 +71,7 @@ public class PlanNewCommonOutController {
     private PlanServiceInterface planServiceInterface;
     private static final ConcurrentHashMap<String, Integer> MAP = new ConcurrentHashMap<>();
 
-    // @Scheduled(cron = "1 1 4 * * ?")
-    // @SchedulerLock(name = "launch:PlanNewCommonOutController:clearUserContextHashMap", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
-    // public void clearUserContextHashMap() {
-    //     log.info("清空计划-H5用户Map数据》》》》》》》》》》》》》》》》》》》》》》开始");
-    //     MAP.clear();
-    //     log.info("清空计划-H5用户Map数据》》》》》》》》》》》》》》》》》》》》》》 结束");
-    // }
+
 
     /**
      * 通过 code 获取计划配置H5信息
@@ -83,7 +80,7 @@ public class PlanNewCommonOutController {
      * @return {@link Mono}<{@link RStatus}<{@link PlanNewCommon}>>
      */
     @GetMapping("getPlanNewCommonH5")
-    public Mono<RStatus<byte[]>> getPlanNewCommonH5(@RequestParam("code") String code) {
+    public Mono<RStatus<byte[]>> getPlanNewCommonH5(@RequestParam("code") String code, ServerWebExchange exchange) {
         Mono<PlanNewCommon> planNewCommonMono = template.selectOne(Query.query(Criteria.where("code").is(code)), PlanNewCommon.class)
                 .switchIfEmpty(Mono.error(new BaseException("计划code异")))
                 .onErrorResume(throwable -> Mono.error(new BaseException("计划code异")));
@@ -132,11 +129,29 @@ public class PlanNewCommonOutController {
                     Image image = imageList.get(RandomUtil.randomInt(0, imageList.size()));
                     h5ImageAndText.setImage(image.image());
                 }
-
                 List<Material> materialList = tuple.getT4();
                 if (!materialList.isEmpty()) {
                     Material material = materialList.get(RandomUtil.randomInt(0, materialList.size()));
-                    h5ImageAndText.setName(material.content());
+                    String content = material.content();
+                    String clientIP=null ;//exchange.getRequest().getHeaders().getFirst("X-Forwarded-For");
+                    if ( Objects.nonNull(exchange.getRequest().getRemoteAddress())) {
+                        clientIP = exchange.getRequest().getRemoteAddress().getAddress().getHostAddress();
+                    }
+                    if (StrUtil.isNotBlank(clientIP)) {
+                        JSONObject ipInfo = ipAnalyzeServiceInterface.getIpInfo(clientIP);
+                        if (null != ipInfo) {
+                            String city = ipInfo.getStr("city");
+                            if (StrUtil.isNotBlank(city)) {
+                                content = city + content;
+                            } else {
+                                String province = ipInfo.getStr("region");
+                                if (StrUtil.isNotBlank(province)) {
+                                    content = province + content;
+                                }
+                            }
+                        }
+                    }
+                    h5ImageAndText.setName(content);
                     h5ImageAndText.setRemark(material.remark());
                 }
                 h5ImageAndTextList.add(h5ImageAndText);
@@ -173,6 +188,9 @@ public class PlanNewCommonOutController {
      */
     @PostMapping("kefuList")
     public Mono<RStatus<byte[]>> kefuList(@RequestBody KefuRequest request) {
+        if (null == request.getCode() || request.getCode().isEmpty()) {
+            return Mono.error(new BaseException("code为空"));
+        }
         Mono<PlanNewCommon> code = template.selectOne(Query.query(Criteria.where("code").is(request.getCode())), PlanNewCommon.class)
                 .switchIfEmpty(Mono.error(new BaseException("code查询返回结构为空")))
                 .onErrorResume(throwable -> Mono.error(new BaseException("code查询返回结构为空")));
@@ -262,6 +280,9 @@ public class PlanNewCommonOutController {
      */
     @PostMapping("domainSelect")
     public Mono<RStatus<DomainSelectRes>> domainSelect(@RequestBody OutDomainRequest request) {
+        if (null == request.getCode() || request.getCode().isEmpty()) {
+            return Mono.error(new BaseException("code为空"));
+        }
         if (1 == request.getTypePlan()) {
             // 加群裂变
             return this.getPlanNew(request);
@@ -496,6 +517,9 @@ public class PlanNewCommonOutController {
      */
     @PostMapping("getCommonMaterial")
     public Mono<RStatus<PlanNewCommonMaterialRes>> getCommonMaterial(@RequestBody PlanNewCommonMaterialReq request) {
+        if (null == request.getCode() || request.getCode().isEmpty()) {
+            return Mono.error(new BaseException("code为空"));
+        }
         Mono<PlanNewCommon> planNewCommonMono = template.selectOne(Query.query(Criteria.where("code").is(request.getCode())), PlanNewCommon.class)
                 .switchIfEmpty(Mono.error(new BaseException("计划code异常 返回结果为空")))
                 .onErrorResume(throwable -> Mono.error(new BaseException("计划code异常")));
@@ -684,7 +708,12 @@ public class PlanNewCommonOutController {
         // Mono<byte[]> res = resMono1.flatMap(f -> Mono.just(Objects.requireNonNull(this.compressData(JSONUtil.toJsonStr(f)))));
         return RStatus.successList(resMono1);
     }
-
+    private QrcodeInfo getQrcodeInfo(GoRequest goRequest)  {
+        Optional<Item> first = goRequest.getItem().stream().filter(ff -> "cover".equals(ff.getT()) ).findFirst();
+        Optional<Item> first3 = goRequest.getItem().stream().filter(ff -> "title".equals(ff.getT())).findFirst();
+        QrcodeInfo qrcodeInfo = new QrcodeInfo(null,first.isPresent()?first.get().getV():"",first3.isPresent()?first3.get().getV():"");
+        return qrcodeInfo;
+    }
     /**
      * 设置二维码 生成图片
      */
@@ -704,34 +733,60 @@ public class PlanNewCommonOutController {
                 PlanNewBackgroupPoster fxSetting = fm.getFxSetting();
                 goRequest = setNameVal(posterTemplate, fxSetting, fm.getSpare());
             }
-            String ret;
-            if (Objects.nonNull(goRequest)) {
-                ret = goGenerateService.goGenerate(JSONUtil.toJsonStr(goRequest));
-            } else {
-
-                ret = goGenerateService.goGenerate(posterTemplate);
+            QrcodeInfo stringBuilder = getQrcodeInfo(goRequest);
+            Mono<QrcodeInfo> qrcodeInfoMono=Mono.just(new QrcodeInfo(null,null,null));
+            List<Item> collect = goRequest.getItem()
+                    .stream()
+                    .filter(filter -> "cusqrcode".equals(filter.getT())
+                            && filter.getIsRandom()
+                            && 1 == filter.getType()).collect(Collectors.toList());
+            if (!collect.isEmpty()) {
+                 qrcodeInfoMono = template.insert(stringBuilder)
+                        .onErrorResume(throwable -> Mono.error(new BaseException("插入QrcodeInfo异常：：：：" + throwable.getMessage())));
             }
-            if (Objects.nonNull(ret)) {
-                JSONObject jsonObject1;
-                try {
-                    jsonObject1 = JSONUtil.parseObj(ret);
-                } catch (Exception e) {
-                    return Mono.error(new BaseException(e.getMessage() + "---" + JSONUtil.parseObj(goRequest) + " error:" + LogExceptionStackTrace.erroStackTrace(e).toString()));
-                }
-                if (jsonObject1.containsKey("code") && jsonObject1.get("code", Integer.class) == 0 && jsonObject1.containsKey("data")) {
-                    JSONObject data = jsonObject1.get("data", JSONObject.class);
-                    if (data.containsKey("Url")) {
-                        String url = data.get("Url", String.class);
-                        if (Objects.nonNull(request.getWx_user_id()) && !request.getWx_user_id().isEmpty() && 3 != fm.getPlanNew().type()) {
-                            fm.setQsGenerateImagesUrl(url);
-                        } else {
-                            fm.setFxGenerateImagesUrl(url);
+
+          return   Mono.zip(Mono.just(fm),qrcodeInfoMono).flatMap(fm1->{
+                if (Objects.nonNull(fm1.getT2().id())) {
+                    goRequest.getItem()
+                            .stream()
+                            .filter(filter -> "cusqrcode".equals(filter.getT())
+                                    && filter.getIsRandom()
+                                    && 1 == filter.getType()).collect(Collectors.toList()).forEach(fe -> {
+                        if (Objects.nonNull(fe.getV())) {
+                            if (fe.getV().contains("?")){
+                                String cusqrcode = fe.getV() + "&id=" + fm1.getT2().id();
+                                fe.setV(cusqrcode);
+                            }else{
+                                String cusqrcode = fe.getV() + "?id=" + fm1.getT2().id();
+                                fe.setV(cusqrcode);
+                            }
+
                         }
+                    });
+                }
+                String  ret = goGenerateService.goGenerate(JSONUtil.toJsonStr(goRequest));
+                if (Objects.nonNull(ret)) {
+                    JSONObject jsonObject1;
+                    try {
+                        jsonObject1 = JSONUtil.parseObj(ret);
+                    } catch (Exception e) {
+                        throw new BaseException(e.getMessage() + "---" + JSONUtil.parseObj(goRequest));
+                    }
+                    if (jsonObject1.containsKey("code") && jsonObject1.get("code", Integer.class) == 0 && jsonObject1.containsKey("data")) {
+                        JSONObject data = jsonObject1.get("data", JSONObject.class);
+                        if (data.containsKey("Url")) {
+                            String url = data.get("Url", String.class);
+                            if (Objects.nonNull(request.getWx_user_id()) && !request.getWx_user_id().isEmpty() && 3 != fm.getPlanNew().type()) {
+                                fm1.getT1().setQsGenerateImagesUrl(url);
+                            } else {
+                                fm1.getT1().setFxGenerateImagesUrl(url);
+                            }
 
+                        }
                     }
                 }
-            }
-            return Mono.just(fm);
+                return Mono.just(fm);
+            });
         });
         return resMono1;
     }

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

@@ -670,7 +670,7 @@ public class PlanNewOutController {
             }
             QrcodeInfo stringBuilder = getQrcodeInfo(goRequest);
             Mono<QrcodeInfo> qrcodeInfoMono = template.insert(stringBuilder)
-                    .onErrorResume(throwable -> Mono.error(new BaseException("插入QrcodeInfo异常：：：：" + throwable.getMessage())));;
+                    .onErrorResume(throwable -> Mono.error(new BaseException("插入QrcodeInfo异常：：：：" + throwable.getMessage())));
 //            if(Objects.nonNull(fm.getPlanNew().groupListDisplay()) && fm.getPlanNew().groupListDisplay()==1){
 //                qrcodeInfoMono = template.insert(stringBuilder)
 //                                .onErrorResume(throwable -> Mono.error(new BaseException("插入QrcodeInfo异常：：：：" + throwable.getMessage())));

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/service/listening/ListeningPlanServiceImpl.java

@@ -24,6 +24,8 @@ import com.webflux.launchadmin.mysql.service.planNew.PlanServiceInterface;
 import com.webflux.launchadmin.mysql.service.planNew.goGenerate.GoGenerateService;
 import com.webflux.launchadmin.mysql.service.planNew.structure.GoRequest;
 import com.webflux.launchadmin.mysql.service.planNew.structure.Item;
+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.RStatus;
 import jakarta.annotation.Resource;
@@ -55,6 +57,10 @@ public class ListeningPlanServiceImpl implements ListeningPlanServiceInterface{
     private R2dbcEntityTemplate template;
     @Resource
     private PlanServiceInterface planServiceInterface;
+    @Resource
+    private SyncListeningAudioToRedis syncListeningAudioToRedis;
+    @Resource
+    private SyncPlanNewListenToRedis syncPlanNewListenToRedis;
     @Override
     public Mono<RStatus<Paged<ListeningPlanNewRes2>>> page(PlanReqPage request, Criteria criteria) {
         Query query = Query.query(criteria);
@@ -273,7 +279,11 @@ public class ListeningPlanServiceImpl implements ListeningPlanServiceInterface{
         } else {
            updateOrinsert = template.insert(listeningAudio);
         }
-        return RStatus.success(updateOrinsert);
+        Mono<ListeningAudio> listeningAudioMono = updateOrinsert.flatMap(fm -> {
+            syncListeningAudioToRedis.setRedisValue(fm).subscribe();
+            return Mono.just(fm);
+        });
+        return RStatus.success(listeningAudioMono);
     }
 
     @Override
@@ -383,14 +393,18 @@ public class ListeningPlanServiceImpl implements ListeningPlanServiceInterface{
 
     @Override
     public Mono<RStatus<ListeningDataLists>> LandingPageList(OutListeningRequest outListeningRequest) {
-        Mono<ListeningPlanNew> code1 = template.selectOne(Query.query(Criteria.where("code")
-                .is(outListeningRequest.getCode())), ListeningPlanNew.class)
+//        Mono<ListeningPlanNew> code1 = template.selectOne(Query.query(Criteria.where("code")
+//                .is(outListeningRequest.getCode())), ListeningPlanNew.class)
+//                .switchIfEmpty(Mono.error(new BaseException(" code错误")))
+//                .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())));
+        Mono<ListeningPlanNew> code1 = syncPlanNewListenToRedis.byRedisGetPlanNewIfErByMysql(outListeningRequest.getCode())
                 .switchIfEmpty(Mono.error(new BaseException(" code错误")))
                 .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())));
         Mono<ListeningDataLists> idx_num1 = code1.flatMap(f -> {
-            Mono<List<ListeningAudio>> idx_num = template.select(Query.query(Criteria.empty())
-                    .sort(Sort.by(Sort.Order.asc("idx_num")))
-                    .limit(f.pageShow()), ListeningAudio.class).collectList();
+            Mono<List<ListeningAudio>> idx_num =syncListeningAudioToRedis.byRedisGetLinteningAudioIfErByMysql(f.pageShow());
+//            Mono<List<ListeningAudio>> idx_num = template.select(Query.query(Criteria.empty())
+//                    .sort(Sort.by(Sort.Order.asc("idx_num")))
+//                    .limit(f.pageShow()), ListeningAudio.class).collectList();
             return Mono.zip(Mono.just(f), idx_num).flatMap(ff -> {
                 ListeningDataLists listeningDataLists = new ListeningDataLists();
                 ListeningPlanNew t1 = ff.getT1();
@@ -496,6 +510,21 @@ public class ListeningPlanServiceImpl implements ListeningPlanServiceInterface{
         return RStatus.success(dataListsShareMono);
     }
 
+    @Override
+    public Mono<RStatus<Long>> deleteAudioById(Long newAudioId) {
+        Query new_audio_id = Query.query(Criteria.where("new_audio_id")
+                .is(newAudioId));
+        Mono<ListeningAudio> listeningAudioMono = template.selectOne(new_audio_id, ListeningAudio.class)
+                .switchIfEmpty(Mono.error(new BaseException("待删除的数据不存在 请刷新页面")))
+                .flatMap(fm -> template.delete(fm));
+        Mono<Long> longMono = listeningAudioMono.flatMap(f -> {
+            syncListeningAudioToRedis.deleteRedis(f).subscribe();
+            return Mono.just(f.newAudioId());
+        });
+        return RStatus.success(longMono);
+
+    }
+
     private void setQrcodeV(reactor.util.function.Tuple5<ListeningPlanNew, PosterTemplate, List<Poster>, List<ListeningPlanNewBackgroupPoster>, StringBuilder> f, GoRequest goRequest, ListeningPlanNewBackgroupPoster fxSetting) {
         if (Objects.nonNull(fxSetting)) {
             Optional<Item> first = goRequest.getItem().stream().filter(fz -> "qrcode".equals(fz.getT())).findFirst();

launch-admin/src/main/java/com/webflux/launchadmin/mysql/service/listening/ListeningPlanServiceInterface.java

@@ -31,4 +31,5 @@ public interface ListeningPlanServiceInterface {
     Mono<RStatus<String>> copyById(String listeningPlanNewId);
     Mono<RStatus<ListeningDataLists>> LandingPageList(OutListeningRequest outListeningRequest);
     Mono<RStatus<DataListsShare>> share(OutListeningShareRequest outListeningRequest);
+    Mono<RStatus<Long>> deleteAudioById(Long newAudioId);
 }

launch-admin/src/main/java/com/webflux/launchadmin/mysql/service/planNew/goGenerate/GoGenerateService.java

@@ -599,7 +599,11 @@ public class GoGenerateService {
 //            }else{
 //                goRequest.getItem().stream().filter(fc->"cusqrcode".equals(fc.getT())).forEach(fe->fe.setName("客服二维码"));
 //            }
-            Optional<Item> first2 = goRequest.getItem().stream().filter(fc -> "cusqrcode".equals(fc.getT()) && (Objects.isNull(fc.getV()) || "".equals(fc.getV()))).findFirst();
+            Optional<Item> first2 = goRequest.getItem()
+                    .stream()
+                    .filter(fc -> "cusqrcode".equals(fc.getT())
+                            && (Objects.isNull(fc.getV())
+                            || "".equals(fc.getV()))).findFirst();
             if (first2.isPresent()) {
                 throw new BaseException("自定义二维码素材为空");
             }
@@ -752,8 +756,25 @@ public class GoGenerateService {
             Mono<QrcodeInfo> qrcodeInfoMono =template.insert(stringBuilder)
                     .switchIfEmpty(Mono.error(new BaseException("插入异常返回结果为空")))
                     .onErrorResume(throwable -> Mono.error(new BaseException("插入QrcodeInfo异常：：：：" + throwable.getMessage())));
+
+
           return Mono.zip(qrcodeInfoMono,Mono.just(goRequest)).flatMap(fff->{
-                GoRequest t21 = fff.getT2();
+              GoRequest t21 = fff.getT2();
+               t21.getItem()
+                      .stream()
+                      .filter(filter -> "cusqrcode".equals(filter.getT())
+                              && filter.getIsRandom()
+                              && 1==filter.getType()).forEach(fe -> {
+                   if (Objects.nonNull(fe.getV())) {
+                       if (fe.getV().contains("?")){
+                           String cusqrcode = fe.getV() + "&id=" + fff.getT1().id();
+                           fe.setV(cusqrcode);
+                       }else{
+                           String cusqrcode = fe.getV() + "?id=" + fff.getT1().id();
+                           fe.setV(cusqrcode);
+                       }
+                   }
+               });
                 t21.getItem().stream().filter(f1 -> "qrcode".equals(f1.getT())).forEach(fe->{
                     if (Objects.nonNull(fe.getV())) {
                         if (fe.getV().contains("?")){
@@ -1232,17 +1253,19 @@ public class GoGenerateService {
                 } else {
                     throw new BaseException("type错误::" + type);
                 }
-                Query query = Query.query(Criteria.where(column).is(planId)).columns("group_id", "code");
+
                 Mono<StringBuilder> stringMono;
                 if (1 == type) {
+                    Query query = Query.query(Criteria.where(column).is(planId)).columns("group_id", "code","import_domain");
                     stringMono = template.selectOne(query, PlanNew.class)
                             .switchIfEmpty(Mono.error(new BaseException("自定义二维码:裂变计划查询::PlanId 返回空结果::" + planId)))
                             .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())))
                             .flatMap(f1 -> {
                                 String code = f1.code();
-                                return getDomainSetV(ftm, planId, type, f1.groupId(), code);
+                                return getDomainSetV(ftm, planId, type, Long.valueOf(f1.importDomain()), code);
                             });
                 } else if (2 == type) {
+                    Query query = Query.query(Criteria.where(column).is(planId)).columns("group_id", "code");
                     stringMono = template.selectOne(query, ListeningPlanNew.class)
                             .switchIfEmpty(Mono.error(new BaseException("自定义二维码:听书计划查询::PlanId  返回空结果::" + planId)))
                             .onErrorResume(throwable -> Mono.error(new BaseException("select ListeningPlanNew::" + throwable.getMessage())))
@@ -1251,12 +1274,13 @@ public class GoGenerateService {
                                 return getDomainSetV(ftm, planId, type, f1.groupId(), code);
                             });
                 } else {
+                    Query query = Query.query(Criteria.where(column).is(planId)).columns("group_id", "code","import_domain");
                     stringMono = template.selectOne(query, PlanNewCommon.class)
                             .switchIfEmpty(Mono.error(new BaseException("自定义二维码:交友计划查询::PlanId 返回空结果::" + planId)))
                             .onErrorResume(throwable -> Mono.error(new BaseException(throwable.getMessage())))
                             .flatMap(f1 -> {
                                 String code = f1.code();
-                                return getDomainSetV(ftm, planId, type, f1.groupId(), code);
+                                return getDomainSetV(ftm, planId, type, Long.valueOf(f1.importDomain()), code);
                             });
                 }
                 return stringMono;

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

@@ -0,0 +1,233 @@
+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.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 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 */30 * * * ?")
+    @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.opsForHash().put(LinteningAudioRedisKey.lintening_Audio_INFO_HASH , stringBuilder, json).subscribe();
+        reactiveRedisTemplate.opsForSet().add(LinteningAudioRedisKey.PLAN_NEW_INFO_SET,stringBuilder ).subscribe();
+       return Mono.empty();
+    }
+    public Mono<Void> deleteAll() {
+        deleteByPattern("sync_image_*").subscribe();
+        deleteByPattern("sync_lintening_*").subscribe();
+        deleteByPattern("sync_plan_new_*").subscribe();
+     //   reactiveRedisTemplate.delete("sync_lintening_*").subscribe();
+        return Mono.empty();
+    }
+    public Mono<Long> deleteByPattern(String pattern) {
+        return reactiveRedisTemplate
+                .scan(ScanOptions.scanOptions().match(pattern).build())
+                .collectList()
+                .flatMap(keys -> {
+                    if (keys.isEmpty()) {
+                        return Mono.just(0L);
+                    }
+                    return reactiveRedisTemplate.delete(Flux.fromIterable(keys));
+                });
+    }
+
+    @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.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();
+        return Mono.empty();
+    }
+
+    /**
+     * 通过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()))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis ListeningAudio 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return 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
+                              .opsForHash()
+                              .get(LinteningAudioRedisKey.lintening_Audio_INFO_HASH , f))
+                              .flatMap(str -> {
+                                  if (Objects.nonNull(str) && !"".equals(str)) {
+                                    return   Mono.just(strParsePlanNew(String.valueOf(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,200 @@
+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.Objects;
+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 */29 * * * ?")
+    @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");
+    }
+
+
+    @NotNull
+    public Mono<Void> setRedisValue(ListeningPlanNew planNew) {
+        reactiveRedisTemplate.opsForSet().add(LinteningRedisKey.PLAN_NEW_INFO_SET, planNew.code()).subscribe();
+        String json= getPlanNewToString(planNew);
+        reactiveRedisTemplate.opsForHash().put(LinteningRedisKey.PLAN_NEW_INFO_HASH,planNew.code(), json).subscribe();
+       return Mono.empty();
+    }
+    @NotNull
+    public Mono<Void> getDelete(String code) {
+        reactiveRedisTemplate.opsForSet().remove(LinteningRedisKey.PLAN_NEW_INFO_SET, code).subscribe();
+        reactiveRedisTemplate.opsForHash().remove(LinteningRedisKey.PLAN_NEW_INFO_HASH , code).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());
+        }
+    }
+
+
+
+
+
+    /**
+     * 通过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.opsForHash()
+                 .get(LinteningRedisKey.PLAN_NEW_INFO_HASH , code)
+                 .switchIfEmpty(Mono.just(""))
+              .onErrorResume(throwable -> {
+                  log.error("接口异常  {} e = {}","获取redis ListeningPlanNew 失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(throwable));
+                  return Mono.just("");
+              })
+              .flatMap(s -> {
+                     Mono<ListeningPlanNew> planNewMono;
+                     if(!"".equals(s) && Objects.nonNull(s)){
+                         try {
+                             ListeningPlanNew planNew = strParsePlanNew(String.valueOf(s));
+                             planNewMono=Mono.just(planNew);
+                         }catch (Exception e){
+                             log.error("接口异常  {} e = {}","获取redis Value 解析ListeningPlanNew失败 暂时从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

@@ -0,0 +1,196 @@
+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';
+     */
+      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:SyncPlanNewToRedis:syncFissionPlan", 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());
+                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()) {
+                     Flux.fromIterable(removeIds).flatMap(this::getDelete).subscribe();
+                }
+              return  Mono.empty();
+            }).then();
+        }).subscribe();
+    }
+
+    @NotNull
+    public Mono<Void> setRedisValue(PlanNew planNew) {
+        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();
+       return Mono.empty();
+    }
+
+    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) {
+       reactiveRedisTemplate.opsForSet().remove(FissionRedisKey.PLAN_NEW_INFO_SET, code).subscribe();
+       reactiveRedisTemplate.opsForValue().delete(FissionRedisKey.PLAN_NEW_INFO_KEY + code).subscribe();
+       return Mono.empty();
+    }
+
+    /**
+     * 通过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());
+        }
+    }
+}

launch-admin/src/main/resources/application-prod.yml

@@ -9,6 +9,7 @@ spring:
       jedis:
         pool:
           max-active: 1000
+          max-wait: 2000ms # 最大等待时间，单位是毫秒
   r2dbc:
     url: r2dbcs:mysql://rm-uf6xylut5n33hb3o5.mysql.rds.aliyuncs.com:3306/wx_share_8?serverZoneId=GMT%2b8&useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&ssl=false&sslMode=DISABLED
     username: wx_share_liuwei

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

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

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

@@ -0,0 +1,17 @@
+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_";
+    public final static String lintening_Audio_INFO_HASH = "sync_lintening_Audio_info_hash_json";
+}

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

@@ -0,0 +1,17 @@
+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_";
+    public final static String PLAN_NEW_INFO_HASH = "sync_lintening_plan_new_info_hash_json";
+}