裂变计划redis v0.04 +听书

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

@@ -20,6 +20,7 @@ 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;
@@ -52,7 +53,7 @@ public class SyncListeningAudioToRedis {
     /**
      *同步裂变计划
      */
-    @Scheduled(cron="0 */1 * * * ?")
+    @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())
@@ -91,12 +92,28 @@ public class SyncListeningAudioToRedis {
     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.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) {
@@ -121,14 +138,14 @@ public class SyncListeningAudioToRedis {
     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();
+        reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.lintening_Audio_INFO_HASH, 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();
+         reactiveRedisTemplate.opsForSet().remove(LinteningAudioRedisKey.PLAN_NEW_INFO_SET, key).subscribe();
+         reactiveRedisTemplate.opsForHash().remove(LinteningAudioRedisKey.lintening_Audio_INFO_HASH, key).subscribe();
+        return Mono.empty();
     }
 
     /**
@@ -169,11 +186,11 @@ public class SyncListeningAudioToRedis {
                               .limit(size)
                               .collect(Collectors.toList());
                       return Flux.fromIterable(collect).flatMap(f -> reactiveRedisTemplate
-                              .opsForValue()
-                              .get(LinteningAudioRedisKey.PLAN_NEW_INFO_KEY + f))
+                              .opsForHash()
+                              .get(LinteningAudioRedisKey.lintening_Audio_INFO_HASH , f))
                               .flatMap(str -> {
                                   if (Objects.nonNull(str) && !"".equals(str)) {
-                                    return   Mono.just(strParsePlanNew(str));
+                                    return   Mono.just(strParsePlanNew(String.valueOf(str)));
                                   }
                                   return Mono.empty();
                               }).collectList();

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

@@ -33,6 +33,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
@@ -50,7 +51,7 @@ public class SyncPlanNewListenToRedis {
     /**
      *同步裂变计划
      */
-    @Scheduled(cron="0 */1 * * * ?")
+    @Scheduled(cron="0 */29 * * * ?")
     @SchedulerLock(name = "launch:SyncPlanNewListenToRedis:syncLinteningPlan", lockAtLeastFor = "PT5S", lockAtMostFor = "PT15M")
     public void syncLinteningPlan(){
         System.out.println("start");
@@ -87,22 +88,21 @@ public class SyncPlanNewListenToRedis {
         }).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();
+        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();
@@ -118,12 +118,7 @@ public class SyncPlanNewListenToRedis {
 
 
 
-    @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
@@ -148,8 +143,8 @@ public class SyncPlanNewListenToRedis {
      * @return
      */
     public Mono<ListeningPlanNew> byRedisGetPlanNewIfErByMysql(String code){
-      return   reactiveRedisTemplate.opsForValue()
-                 .get(LinteningRedisKey.PLAN_NEW_INFO_KEY + 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));
@@ -157,9 +152,9 @@ public class SyncPlanNewListenToRedis {
               })
               .flatMap(s -> {
                      Mono<ListeningPlanNew> planNewMono;
-                     if(!"".equals(s)){
+                     if(!"".equals(s) && Objects.nonNull(s)){
                          try {
-                             ListeningPlanNew planNew = strParsePlanNew(s);
+                             ListeningPlanNew planNew = strParsePlanNew(String.valueOf(s));
                              planNewMono=Mono.just(planNew);
                          }catch (Exception e){
                              log.error("接口异常  {} e = {}","获取redis Value 解析ListeningPlanNew失败 暂时从Mysql获取内容" , LogExceptionStackTrace.erroStackTrace(e));