class RealTimeRedisDataStorage extends AbstractRealTimeDataStorage {

    private RedisClient redisClient;
    private final Integer db;
    private StatefulRedisConnection<String, CollectRep.MetricsData> connection;

    public RealTimeRedisDataStorage(WarehouseProperties properties) {
        this.serverAvailable = initRedisClient(properties);
        this.db = getRedisSelectDb(properties);
    }

    private Integer getRedisSelectDb(WarehouseProperties properties){
        return properties.getStore().getRedis().db();
    }

    @Override
    public CollectRep.MetricsData getCurrentMetricsData(@NonNull Long monitorId, @NonNull String metric) {
        RedisCommands<String, CollectRep.MetricsData> commands = connection.sync();
        commands.select(db);
        return commands.hget(String.valueOf(monitorId), metric);
    }

    @Override
    public List<CollectRep.MetricsData> getCurrentMetricsData(@NonNull Long monitorId) {
        RedisCommands<String, CollectRep.MetricsData> commands = connection.sync();
        commands.select(db);
        Map<String, CollectRep.MetricsData> metricsDataMap = commands.hgetall(String.valueOf(monitorId));
        return new ArrayList<>(metricsDataMap.values());
    }

    @Override
    public void saveData(CollectRep.MetricsData metricsData) {
        String key = String.valueOf(metricsData.getId());
        String hashKey = metricsData.getMetrics();
        if (metricsData.getCode() != CollectRep.Code.SUCCESS || !isServerAvailable()) {
            return;
        }
        if (metricsData.getValuesList().isEmpty()) {
            log.info("[warehouse redis] redis flush metrics data {} - {} is null, ignore.", key, hashKey);
            return;
        }
        RedisAsyncCommands<String, CollectRep.MetricsData> commands = connection.async();
        commands.select(db);
        commands.hset(key, hashKey, metricsData).thenAccept(response -> {
            if (response) {
                log.debug("[warehouse] redis add new data {}:{}.", key, hashKey);
            } else {
                log.debug("[warehouse] redis replace data {}:{}.", key, hashKey);
            }
        });
    }

    private boolean initRedisClient(WarehouseProperties properties) {<FILL_FUNCTION_BODY>}

    @Override
    public void destroy() {
        if (connection != null) {
            connection.close();
        }
        if (redisClient != null) {
            redisClient.shutdown();
        }
    }
}

        if (properties == null || properties.getStore() == null || properties.getStore().getRedis() == null) {
            log.error("init error, please config Warehouse redis props in application.yml");
            return false;
        }
        WarehouseProperties.StoreProperties.RedisProperties redisProp = properties.getStore().getRedis();
        RedisURI.Builder uriBuilder = RedisURI.builder()
                .withHost(redisProp.host())
                .withPort(redisProp.port())
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS));
        if (redisProp.password() != null && !"".equals(redisProp.password())) {
            uriBuilder.withPassword(redisProp.password().toCharArray());
        }
        try {
            redisClient = RedisClient.create(uriBuilder.build());
            connection = redisClient.connect(new MetricsDataRedisCodec());
            return true;
        } catch (Exception e) {
            log.error("init redis error {}", e.getMessage(), e);
        }
        return false;
    
<methods>public non-sealed void <init>() ,public abstract org.apache.hertzbeat.common.entity.message.CollectRep.MetricsData getCurrentMetricsData(java.lang.Long, java.lang.String) ,public abstract List<org.apache.hertzbeat.common.entity.message.CollectRep.MetricsData> getCurrentMetricsData(java.lang.Long) ,public boolean isServerAvailable() <variables>protected boolean serverAvailable