ROSBOX-软路由盒子之家

设计

首页 > 设计
12/13 2023
By javacoo

Vert.x 响应式微服务设计之-缓存设计

设计思路

思路很简单，无非就是优先使用缓存数据，如果缓存为空，则从数据库加载数据，并缓存，但是由于我们是使用Vert.x实现响应式的微服务，所以上述步骤必须以异步的方式实现，整合这个异步的过程将是一个难点，这里我们使用到了响应式数据库客户端：vertx-jdbc-client和响应式Redis客户端：vertx-redis-client，它们提供了丰富的API，能够帮助我们轻松实现上诉目标。

响应式数据库客户端：vertx-jdbc-client，主要用到了基于RxJava版本的相关API,如：rxQueryWithParams，基于RxJava版本的API,利用了RxJava的发布订阅机制，在写法上更简洁，直观，优雅，避免了原生版本的回调地狱等缺点，大大提高了代码的可读性和可维护性。

响应式Redis客户端：vertx-redis-client，主要用到了三个API：get,getset,expire，分表表示：获取缓存数据，设置缓存数据，设置缓存KEY过期时间。

核心交互过程如下：

  1. 读取缓存

//首先从缓存中获取值，如果没有值，则查询数据库，并缓存，否则使用缓存的值。
        RedisAPIHolder.getRedisAPI().get(cacheKey)
                .onSuccess(val -> {
                    //缓存中没有数据，则从数据加载，并缓存
                    if(val == null){
                        LOGGER.info("预编译SQL:{},参数:{}",finalSql,dataAll);
                        //被观察者对象
                        Single<ResultSet> resultSet = dbClient.rxQueryWithParams(finalSql, dataAll);
                        //观察者对象
                        SingleObserver<List<JsonObject>> singleObserver = RedisAPIHolder.toObserver(cacheKey, resultHandler);
                        //绑定：观察者对象订阅被观察者对象发布的事件
                        resultSet.map(ResultSet::getRows).subscribe(singleObserver);
                    }else{//缓存有值，则使用缓存的值
                        LOGGER.info("从缓存获取值:key->{}", cacheKey);
                        List<JsonObject> jsonObjectList = toJsonObjectList(val.toString());
                        resultHandler.handle(Future.succeededFuture(jsonObjectList));
                    }
                })
                .onFailure(event -> LOGGER.error("从缓存获取值失败:key->{},{}", cacheKey,event.getCause()));


  • 缓存数据并设置过期时间

    • /**
     * 异步结果观察者对象
     * <li></li>
     * @author javacoo
     * @date 2023/6/11 22:25

     * @param cacheKey 缓存KEY
     * @param handler 异步结果处理器
     * @return: SingleObserver<T>
     */
    public static <T> SingleObserver<T> toObserver(String cacheKey,Handler<AsyncResult<T>> handler) {
        AtomicBoolean completed = new AtomicBoolean();
        return new SingleObserver<T>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }
            @Override
            public void onSuccess(@NonNull T item) {
                if (completed.compareAndSet(false, true)) {
                    //响应结果
                    handler.handle(Future.succeededFuture(item));
                    //缓存数据,并设置过期时间
                    redisAPI.getset(cacheKey, JSONUtil.toJsonStr(item))
                            .onSuccess(val -> {
                                LOGGER.info("查询数据库获取值,并缓存:key->{},expireTime->{}秒",cacheKey,redisProperties.getExpireTime());
                                //设置过期时间
                                List<String> command = new ArrayList<>();
                                command.add(cacheKey);
                                command.add(String.valueOf(redisProperties.getExpireTime()));
                                redisAPI.expire(command);
                            })
                            .onFailure(event -> LOGGER.error("缓存数据失败:key->{},{}",cacheKey,event.getCause()));

                }
            }
            @Override
            public void onError(Throwable error) {
                if (completed.compareAndSet(false, true)) {
                    handler.handle(Future.failedFuture(error));
                }
            }
        };
    }


    相关配置及核心代码实现

    1. 导入依赖

          <dependency>
         <groupId>org.javacoo.vertx</groupId>
         <artifactId>vertx-core</artifactId>
         <version>1.0</version>
     </dependency>
     <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-rx-java2</artifactId>
         <version>${vertx.version}</version>
     </dependency>
     <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-jdbc-client</artifactId>
         <version>${vertx.version}</version>
     </dependency>
     <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-config-yaml</artifactId>
         <version>${vertx.version}</version>
     </dependency>
     <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-pg-client</artifactId>
         <version>${vertx.version}</version>
     </dependency>
     <dependency>
         <groupId>org.postgresql</groupId>
         <artifactId>postgresql</artifactId>
         <version>42.2.2</version>
     </dependency>
     <dependency>
         <groupId>io.vertx</groupId>
         <artifactId>vertx-redis-client</artifactId>
         <version>${vertx.version}</version>
     </dependency>


  • 配置文件

    • redis:
  urls:
    - redis://127.0.0.1:6379/0  clientType: STANDALONE  poolName: p-red  poolCleanerInterval: 30000
  maxPoolSize: 8
  maxPoolWaiting: 32
  password: 123456


  • 配置对象：包含在：ConfigProperties配置对象中

    • public static class RedisProperties {
        private RedisClientType clientType = RedisClientType.STANDALONE;
        private String[] urls;
        private String password;
        private String poolName = "redis-p";
        private int poolCleanerInterval = 30_000;
        private int maxPoolSize = 8;
        private int maxPoolWaiting = 32;
        private String masterName = "mymaster";
        private RedisRole role = RedisRole.MASTER;

        private Integer expireTime = 60 * 60;
    //省略getter,setter
    ...}


  • 加载配置：加载系统相关配置

    • private ConfigRetrieverOptions initOptions() {
        // 使用默认ConfigStore
        ConfigRetrieverOptions options = new ConfigRetrieverOptions().setIncludeDefaultStores(false);
        // 禁用配置刷新
        options.setScanPeriod(-1);
        //加载主配置
        options.addStore(new ConfigStoreOptions()
                .setType("file")
                .setFormat("yaml")
                .setOptional(true)
                .setConfig(new JsonObject().put("path", "application.yaml")));
        String envFile = new StringBuilder("application-").append(env).append(".yaml").toString();
        //加载环境配置
        options.addStore(new ConfigStoreOptions()
                .setType("file")
                .setFormat("yaml")
                .setOptional(true)
                .setConfig(new JsonObject().put("path", envFile)));
        // 禁用缓存
        options.getStores().forEach(store -> {
            store.getConfig().put("cache", "false");
        });
        return options;
    }


  • 初始化系统配置对象

    • public class ConfigPropertiesHolder {
    private static ConfigProperties configProperties;
    public static void init(JsonObject envConfig){
        Objects.requireNonNull(envConfig, "未初始化envConfig");
        configProperties = envConfig.mapTo(ConfigProperties.class);
    }
    public static ConfigProperties getConfigProperties() {
        Objects.requireNonNull(configProperties, "未初始化ConfigProperties");
        return configProperties;
    }}


  • Redis缓存组件：

    初始化

    • //初始化RedisClientHolder
                RedisAPIHolder.init(mainVertx, envConfig, redisConnectionAsyncResult -> {
                    if(redisConnectionAsyncResult.succeeded()){
                        LOGGER.info("redis初始化成功");
                    }else{
                        LOGGER.error("redis初始化失败",redisConnectionAsyncResult.cause());
                    }
                });

    RedisClientHolder：

    /**
 * Redis持有者
 * <li></li>
 *
 * @author javacoo
 * @version 1.0
 * @date 2023/6/11 21:55
 */public class RedisAPIHolder {
    protected static final Logger LOGGER = LoggerFactory.getLogger(RedisAPIHolder.class);
    private static RedisAPI redisAPI;
    private static Redis redis;

    private static ConfigProperties.RedisProperties redisProperties;
    public static void init(Vertx vertx, JsonObject envConfig,Handler<AsyncResult<RedisConnection>> resultHandler) {
        Objects.requireNonNull(envConfig, "未初始化envConfig");
        //解析配置
        redisProperties = envConfig.mapTo(ConfigProperties.class).getRedis();
        RedisOptions options = new RedisOptions()
                .setType(redisProperties.getClientType())
                .setPoolName(redisProperties.getPoolName())
                .setMaxPoolSize(redisProperties.getMaxPoolSize())
                .setMaxPoolWaiting(redisProperties.getMaxPoolWaiting())
                .setPoolCleanerInterval(redisProperties.getPoolCleanerInterval());

        // password
        if (StrUtil.isNotBlank(redisProperties.getPassword())) {
            options.setPassword(redisProperties.getPassword());
        }
        // connect address [redis://localhost:7000, redis://localhost:7001]
        for (String url : redisProperties.getUrls()) {
            options.addConnectionString(url);
        }
        // sentinel
        if (redisProperties.getClientType().equals(RedisClientType.SENTINEL)) {
            options.setRole(redisProperties.getRole()).setMasterName(redisProperties.getMasterName());
        }
        //创建redisclient实例
        redis = Redis.createClient(vertx, options);
        redis.connect(resultHandler);
    }

    public static RedisAPI getRedisAPI() {
        Objects.requireNonNull(redis, "未初始化Redis");
        redisAPI = RedisAPI.api(redis);
        return redisAPI;
    }
    /**
     * 异步结果观察者对象
     * <li></li>
     * @author javacoo
     * @date 2023/6/11 22:25

     * @param cacheKey 缓存KEY
     * @param handler 异步结果处理器
     * @return: SingleObserver<T>
     */
    public static <T> SingleObserver<T> toObserver(String cacheKey,Handler<AsyncResult<T>> handler) {
        AtomicBoolean completed = new AtomicBoolean();
        return new SingleObserver<T>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }
            @Override
            public void onSuccess(@NonNull T item) {
                if (completed.compareAndSet(false, true)) {
                    //响应结果
                    handler.handle(Future.succeededFuture(item));
                    //缓存数据,并设置过期时间
                    redisAPI.getset(cacheKey, JSONUtil.toJsonStr(item))
                            .onSuccess(val -> {
                                LOGGER.info("查询数据库获取值,并缓存:key->{},expireTime->{}秒",cacheKey,redisProperties.getExpireTime());
                                //设置过期时间
                                List<String> command = new ArrayList<>();
                                command.add(cacheKey);
                                command.add(String.valueOf(redisProperties.getExpireTime()));
                                redisAPI.expire(command);
                            })
                            .onFailure(event -> LOGGER.error("缓存数据失败:key->{},{}",cacheKey,event.getCause()));

                }
            }
            @Override
            public void onError(Throwable error) {
                if (completed.compareAndSet(false, true)) {
                    handler.handle(Future.failedFuture(error));
                }
            }
        };
    }}


  • 组件使用

    • /**
     * 加载集合数据
     * <li></li>
     * @author javacoo
     * @date 2023/6/10 22:42
     * @param dbClient dbClient
     * @param resultHandler resultHandler
     * @param finalSql 预编译SQL
     * @param dataAll SQL参数
     * @param cacheKey 缓存KEY
     * @return: void
     */
    public static void loadListData(JDBCClient dbClient,Handler<AsyncResult<List<JsonObject>>> resultHandler, String finalSql, JsonArray dataAll, String cacheKey) {
        //首先从缓存中获取值，如果没有值，则查询数据库，并缓存，否则使用缓存的值。
        RedisAPIHolder.getRedisAPI().get(cacheKey)
                .onSuccess(val -> {
                    //缓存中没有数据，则从数据加载，并缓存
                    if(val == null){
                        LOGGER.info("预编译SQL:{},参数:{}",finalSql,dataAll);
                        //被观察者对象
                        Single<ResultSet> resultSet = dbClient.rxQueryWithParams(finalSql, dataAll);
                        //观察者对象
                        SingleObserver<List<JsonObject>> singleObserver = RedisAPIHolder.toObserver(cacheKey, resultHandler);
                        //绑定：观察者对象订阅被观察者对象发布的事件
                        resultSet.map(ResultSet::getRows).subscribe(singleObserver);
                    }else{//缓存有值，则使用缓存的值
                        LOGGER.info("从缓存获取值:key->{}", cacheKey);
                        List<JsonObject> jsonObjectList = toJsonObjectList(val.toString());
                        resultHandler.handle(Future.succeededFuture(jsonObjectList));
                    }
                })
                .onFailure(event -> LOGGER.error("从缓存获取值失败:key->{},{}", cacheKey,event.getCause()));
    }


  • 测试

    第一次查询：

    • [2023-06-17 21:13:28.126] [level: INFO] [Thread: vert.x-eventloop-thread-1] [ Class:org.javacoo.vertx.core.factory.RouterHandlerFactory >> Method: lambda$createRouter$1:92 ]
INFO:执行交易->appid:null,交易流水号:null,方法:http://127.0.0.1:8331/api/basicEducation/getBasicEduSpecialTeachersTotalByQualificationType,参数:{"areaCode":"510000000000","level":"1"}

[2023-06-17 21:13:28.246] [level: DEBUG] [Thread: vert.x-worker-thread-3] [ Class:io.vertx.redis.client.impl.RedisConnectionManager >> Method: ?:? ]
DEBUG:{mode: standalone, server: redis, role: master, proto: 3, id: 95923, version: 7.0.4, modules: []}

[2023-06-17 21:13:28.344] [level: INFO] [Thread: vert.x-worker-thread-3] [ Class:edu.sc.gis.api.utils.ServiceUtils >> Method: lambda$loadListData$0:40 ]
INFO:预编译SQL:SELECT sum(g.zrjss) total , sum(g.jzgs) empTotal , g.eduType, g.qualificationType FROM(SELECT t1.zrjss, t1.jzgs, replace(t1.xl, 'XL@GJ@','') qualificationType, t.xxbxlx eduType FROM gis_js_xx_jbxx t LEFT JOIN gis_js_xlxb_rs t1 ON t1.gis_xx_jbxx = t.gis_xx_jbxx WHERE t1.zrjss is NOT null AND t1.xl is NOT null AND t.sszgjyxzd LIKE ?) g GROUP BY g.eduType ,g.qualificationType,参数:["51%"]

[2023-06-17 21:13:28.407] [level: DEBUG] [Thread: vert.x-worker-thread-1] [ Class:io.vertx.ext.jdbc.spi.JDBCEncoder >> Method: ?:? ]
DEBUG:Convert JDBC column [JDBCColumnDescriptor[columnName=(0), jdbcTypeWrapper=(JDBCTypeWrapper[vendorTypeNumber=(12), vendorTypeName=(text), vendorTypeClass=(class java.lang.String), jdbcType=(VARCHAR)])]][java.lang.String]

[2023-06-17 21:13:28.980] [level: INFO] [Thread: vert.x-worker-thread-3] [ Class:edu.sc.gis.api.cache.RedisAPIHolder >> Method: lambda$onSuccess$0:83 ]
INFO:查询数据库获取值,并缓存:key->totalTeacherByQueryType_f.qualificationType_51%,expireTime->3600秒

    第二次查询：

    [2023-06-17 21:18:18.121] [level: INFO] [Thread: vert.x-eventloop-thread-1] [ Class:org.javacoo.vertx.core.factory.RouterHandlerFactory >> Method: lambda$createRouter$1:92 ]
INFO:执行交易->appid:null,交易流水号:null,方法:http://127.0.0.1:8331/api/basicEducation/getBasicEduSpecialTeachersTotalByQualificationType,参数:{"areaCode":"510000000000","level":"1"}

[2023-06-17 21:18:18.212] [level: DEBUG] [Thread: vert.x-worker-thread-3] [ Class:io.vertx.redis.client.impl.RedisConnectionManager >> Method: ?:? ]
DEBUG:{mode: standalone, server: redis, role: master, proto: 3, id: 95924, version: 7.0.4, modules: []}

[2023-06-17 21:18:18.392] [level: INFO] [Thread: vert.x-worker-thread-1] [ Class:edu.sc.gis.api.utils.ServiceUtils >> Method: lambda$loadListData$0:44 ]
INFO:从缓存获取值:key->totalTeach


最近更新

RouterOS API简介

2023-12-13T11:46:19

RouterOS脚本语言手册
RouterOS脚本语言手册

2023-12-13T11:48:22

RouterOS防火墙控制两个IP网络段单向访问

2024-01-02T09:07:42

MikroTik RB/CCR/CRS的Reset和Mode按钮使用
MikroTik RB/CCR/CRS的Reset和Mode按钮使用

2024-01-02T09:07:20

RouterOS策略路由导致内网段互访失效(含PCC负载均衡)
RouterOS策略路由导致内网段互访失效(含PCC负载均衡)

2024-01-02T09:06:50

RouterOS v7.1beta3发布
RouterOS v7.1beta3发布

2024-01-02T09:06:26

流量控制的基本原理
流量控制的基本原理

2024-01-02T09:06:01

探讨下RouterOS PCC的工作原理与应用延伸
探讨下RouterOS PCC的工作原理与应用延伸

2024-01-02T09:05:20

SmartDNS的加速和防污染怎么就在国内火了
SmartDNS的加速和防污染怎么就在国内火了

2024-01-02T09:04:49

RouterOS PCC多线比例权重路由
RouterOS PCC多线比例权重路由

2024-01-02T09:04:17