feat: new cache struce such as use redis.

CHANGELOG.MD

@@ -2,12 +2,15 @@
 
 更新日志文档，版本顺序从新到旧，最新版本在最前(上)面。
 
-# 0.19.4
+# 0.20.0
 
 ## 优化
 
-- 移除缓存配置
-- 优化查询剧集附件接口
+- 移除废弃缓存配置
+- 优化剧集查询接口，通过注解套上了缓存
+- 引入redis缓存支持
+- 引入内存缓存支持
+- 引入缓存开关和内存和redis切换配置
 
 # 0.19.3
 

config/server/resource/application-local.yaml.example

@@ -1,19 +1,40 @@
 ikaros:
   show-theme: true
+  enable-redis: true
   security:
     # 30 day for local dev token expire
     jwt-expiration-time: 18144000000
+  indices:
+    initializer:
+      enabled: false
+
+
+
 
   plugin:
     runtime-mode: development
+    auto-start-plugin: true
     fixed-plugin-path:
-    #      - C:\Users\li-guohao\GitRepo\ikaros-dev\plugin-bgmtv
-    #      - C:\Users\li-guohao\GitRepo\ikaros-dev\plugin-alist
-    #      - C:\Develop\GitRepos\ikaros-dev\plugins\plugin-local-files-import
+      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-bgmtv
+    #      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-mikan
+    #      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-alist
+    #      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-starter
+    #      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-local-files-import
+    #      - C:\Users\chivehao\GitRepos\ikaros-dev\plugin-alist
     #      - C:\Develop\GitRepos\ikaros-dev\plugins\plugin-jellyfin
-    #      - C:\Develop\GitRepos\ikaros-dev\plugins\plugin-mikan
     #      - C:\Develop\GitRepos\ikaros-dev\plugins\plugin-baidupan
 
+
+#spring:
+#  r2dbc:
+#    url: r2dbc:pool:postgresql://192.168.13.102:5432/ikaros
+#    username: ikaros
+#    password: openpostgresql
+#  flyway:
+#    url: jdbc:postgresql://192.168.13.102:5432/ikaros
+#    locations: classpath:db/postgresql/migration
+
+
 logging:
   level:
     org.springframework.data.r2dbc: INFO
@@ -24,4 +45,13 @@ logging:
     org.pf4j: INFO
     org.hibernate.SQL: INFO
     org.hibernate.type.descriptor.sql.BasicBinder: INFO
-    org.springframework.r2dbc.core.DefaultDatabaseClient: INFO
+    org.springframework.r2dbc.core.DefaultDatabaseClient: INFO
+
+spring:
+  jpa:
+    show-sql: true
+  data:
+    redis:
+      host: localhost
+      port: 6379
+      password:

gradle.properties

@@ -1 +1 @@
-version=0.19.4
+version=0.20.0

server/build.gradle

@@ -70,6 +70,7 @@ dependencies {
     implementation "org.flywaydb:flyway-core:$flywaydb"
     implementation "io.micrometer:micrometer-registry-prometheus"
 
+    implementation "org.springframework.boot:spring-boot-starter-data-redis-reactive"
 
     implementation 'io.jsonwebtoken:jjwt-api:0.12.5'
     runtimeOnly 'io.jsonwebtoken:jjwt-impl:0.12.5'

server/src/main/java/run/ikaros/server/cache/CacheAspect.java

@@ -0,0 +1,230 @@
+package run.ikaros.server.cache;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.annotation.Pointcut;
+import org.aspectj.lang.reflect.MethodSignature;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import run.ikaros.server.cache.annotation.FluxCacheEvict;
+import run.ikaros.server.cache.annotation.FluxCacheable;
+import run.ikaros.server.cache.annotation.MonoCacheEvict;
+import run.ikaros.server.cache.annotation.MonoCacheable;
+
+@Aspect
+@Component
+@ConditionalOnProperty(value = "ikaros.cache.enable", havingValue = "true")
+public class CacheAspect {
+
+
+    @Pointcut("@annotation(run.ikaros.server.cache.annotation.MonoCacheable) "
+        + "&& execution(public reactor.core.publisher.Mono *(..))")
+    public void monoCacheableMethods() {
+    }
+
+    @Pointcut("@annotation(run.ikaros.server.cache.annotation.FluxCacheable) "
+        + "&& execution(public reactor.core.publisher.Flux *(..))")
+    public void fluxCacheableMethods() {
+    }
+
+    @Pointcut("@annotation(run.ikaros.server.cache.annotation.MonoCacheEvict)")
+    public void monoCacheEvictMethods() {
+    }
+
+    @Pointcut("@annotation(run.ikaros.server.cache.annotation.FluxCacheEvict)")
+    public void fluxCacheEvictMethods() {
+    }
+
+    private final ExpressionParser spelExpressionParser = new SpelExpressionParser();
+    private final ConcurrentHashMap<String, Class<?>> methodReturnValueTypes
+        = new ConcurrentHashMap<>();
+
+    private final ReactiveCacheManager cm;
+
+    public CacheAspect(ReactiveCacheManager cm) {
+        this.cm = cm;
+    }
+
+    /**
+     * 应用关闭时清空缓存.
+     */
+    // @PreDestroy
+    public void onShutdown() throws InterruptedException {
+        // 使用 CountDownLatch 来确保响应式流在退出前执行完成
+        CountDownLatch latch = new CountDownLatch(1);
+
+        cm.clear().then()
+            .doOnTerminate(latch::countDown)  // 当任务完成时计数器减一
+            .subscribe();
+
+        // 等待响应式操作完成
+        latch.await();
+        System.out.println("Shutdown process completed.");
+    }
+
+
+    private String parseSpelExpression(String expression, ProceedingJoinPoint joinPoint) {
+        final EvaluationContext context = new StandardEvaluationContext();
+        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
+        String[] paramNames = methodSignature.getParameterNames();
+        Object[] paramValues = joinPoint.getArgs();
+        for (int i = 0; i < paramNames.length; i++) {
+            context.setVariable(paramNames[i], paramValues[i]);
+        }
+        return spelExpressionParser.parseExpression(expression).getValue(context, String.class);
+    }
+
+    /**
+     * 处理可缓存注解切面
+     * 要求返回值为Mono类型
+     * .
+     */
+    @Around("monoCacheableMethods() && @annotation(monoCacheable)")
+    public Mono<?> aroundMonoMethodsWithAnnotationCacheable(
+        ProceedingJoinPoint joinPoint, MonoCacheable monoCacheable) throws Throwable {
+        final String cacheKeyPostfix = parseSpelExpression(monoCacheable.key(), joinPoint);
+        final List<String> cacheKeys =
+            Arrays.stream(monoCacheable.value())
+                .map(namespace -> namespace + cacheKeyPostfix).toList();
+        return Flux.fromStream(cacheKeys.stream())
+            .concatMap(key -> cm.get(key).filter(Objects::nonNull))
+            .next()
+            // 缓存中不存在
+            .switchIfEmpty(Mono.defer(() -> {
+                Object proceed;
+                try {
+                    proceed = joinPoint.proceed(joinPoint.getArgs());
+                } catch (Throwable e) {
+                    return Mono.error(e);
+                }
+                return ((Mono<?>) proceed)
+                    .flatMap(val ->
+                        Flux.fromIterable(cacheKeys)
+                            .flatMap(k -> cm.put(k, val))
+                            .next()
+                            .flatMap(list -> Mono.just(val))
+                    ).switchIfEmpty(
+                        Flux.fromIterable(cacheKeys)
+                            .flatMap(k -> cm.put(k, "null"))
+                            .next()
+                            .flatMap(bool -> Mono.empty())
+                    );
+            }))
+            .map(o -> {
+                if (o instanceof Integer integer) {
+                    return integer.longValue();
+                }
+                return o;
+            })
+            .filter(o -> !"null".equals(o));
+    }
+
+    /**
+     * 处理可缓存注解切面
+     * 要求返回值为Flux类型
+     * .
+     */
+    @Around("fluxCacheableMethods() && @annotation(fluxCacheable)")
+    public Flux<?> aroundMonoMethodsWithAnnotationCacheable(
+        ProceedingJoinPoint joinPoint, FluxCacheable fluxCacheable) throws Throwable {
+        final String cacheKeyPostfix = parseSpelExpression(fluxCacheable.key(), joinPoint);
+        final List<String> cacheKeys =
+            Arrays.stream(fluxCacheable.value())
+                .map(namespace -> namespace + cacheKeyPostfix).toList();
+        return Flux.fromStream(cacheKeys.stream())
+            .concatMap(key -> cm.get(key)
+                .filter(Objects::nonNull))
+            .next()
+            // 缓存中不存在
+            .switchIfEmpty(Mono.defer(() -> {
+                Object proceed;
+                try {
+                    proceed = joinPoint.proceed(joinPoint.getArgs());
+                } catch (Throwable e) {
+                    throw new RuntimeException(e);
+                }
+
+                return ((Flux<?>) proceed)
+                    .collectList()
+                    .flatMap(vals ->
+                        Flux.fromIterable(cacheKeys)
+                            .flatMap(k -> cm.put(k, vals))
+                            .collectList()
+                            .flatMap(list -> Mono.just(vals))
+                    ).switchIfEmpty(
+                        Flux.fromIterable(cacheKeys)
+                            .flatMap(k -> cm.put(k, List.of()))
+                            .next()
+                            .flatMap(bool -> Mono.empty())
+                    );
+            }))
+            .map(o -> (List<?>) o)
+            .filter(list -> !list.isEmpty())
+            // 缓存中存的是集合
+            .flatMapMany(Flux::fromIterable);
+    }
+
+    /**
+     * 处理缓存移除注解切面
+     * 要求返回值为Mono类型
+     * .
+     */
+    @Around("monoCacheEvictMethods() && @annotation(monoCacheEvict)")
+    public Mono<?> aroundMonoMethodsWithAnnotationCacheable(
+        ProceedingJoinPoint joinPoint, MonoCacheEvict monoCacheEvict
+    ) throws Throwable {
+        Object proceed = joinPoint.proceed();
+        if (monoCacheEvict.value().length == 0
+            && "".equals(monoCacheEvict.key())) {
+            return cm.clear()
+                .flatMap(s -> (Mono<?>) proceed);
+        }
+        final String cacheKeyPostfix = parseSpelExpression(monoCacheEvict.key(), joinPoint);
+        final List<String> cacheKeys =
+            Arrays.stream(monoCacheEvict.value())
+                .map(namespace -> namespace + cacheKeyPostfix).toList();
+        return Flux.fromStream(cacheKeys.stream())
+            .flatMap(cm::remove)
+            .next()
+            .flatMap(bool -> (Mono<?>) proceed);
+    }
+
+    /**
+     * 处理缓存移除注解切面
+     * 要求返回值为Mono类型
+     * .
+     */
+    @Around("fluxCacheEvictMethods() && @annotation(fluxCacheEvict)")
+    public Flux<?> aroundMonoMethodsWithAnnotationCacheable(
+        ProceedingJoinPoint joinPoint, FluxCacheEvict fluxCacheEvict
+    ) throws Throwable {
+        Object proceed = joinPoint.proceed();
+        if (fluxCacheEvict.value().length == 0
+            && "".equals(fluxCacheEvict.key())) {
+            return cm.clear()
+                .flatMapMany(s -> (Flux<?>) proceed);
+        }
+        final String cacheKeyPostfix = parseSpelExpression(fluxCacheEvict.key(), joinPoint);
+        final List<String> cacheKeys =
+            Arrays.stream(fluxCacheEvict.value())
+                .map(namespace -> namespace + cacheKeyPostfix).toList();
+        return Flux.fromStream(cacheKeys.stream())
+            .flatMap(cm::remove)
+            .next()
+            .flatMapMany(bool -> (Flux<?>) proceed);
+    }
+
+
+}

server/src/main/java/run/ikaros/server/cache/CacheConfiguration.java

@@ -0,0 +1,53 @@
+package run.ikaros.server.cache;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import run.ikaros.server.cache.condition.CacheMemoryEnableCondition;
+import run.ikaros.server.cache.condition.CacheRedisEnableCondition;
+
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(CacheProperties.class)
+public class CacheConfiguration {
+
+
+    @Bean
+    @Conditional(CacheMemoryEnableCondition.class)
+    public ReactiveCacheManager memoryReactiveCacheManager() {
+        return new MemoryReactiveCacheManager();
+    }
+
+    @Bean
+    @Conditional(CacheRedisEnableCondition.class)
+    public ReactiveCacheManager redisReactiveCacheManager(
+        ReactiveRedisTemplate<String, Object> reactiveRedisTemplate
+    ) {
+        return new RedisReactiveCacheManager(reactiveRedisTemplate);
+    }
+
+    /**
+     * Redis reactive template.
+     */
+    @Bean
+    @Conditional(CacheRedisEnableCondition.class)
+    public ReactiveRedisTemplate<String, Object> reactiveRedisTemplate(
+        ReactiveRedisConnectionFactory connectionFactory) {
+        RedisSerializationContext.RedisSerializationContextBuilder<String, Object> builder =
+            RedisSerializationContext.newSerializationContext();
+        GenericJackson2JsonRedisSerializer objectSerializer =
+            new GenericJackson2JsonRedisSerializer();
+        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
+        builder.key(stringRedisSerializer);
+        builder.value(objectSerializer);
+        builder.hashKey(stringRedisSerializer);
+        builder.hashValue(objectSerializer);
+        return new ReactiveRedisTemplate<>(connectionFactory, builder.build());
+    }
+
+}