Guava+zookeeper的集群本地缓存小实践

红太狼 2022-05-09 01:40 191阅读 0赞

**目录**

零、依赖

一、结构

二、逻辑

三、代码实现

四、使用

五、问题

六、其他

--------------------

### 零、依赖 ###

<dependency>
    
        <groupId>com.google.guava</groupId>
    
        <artifactId>guava</artifactId>
    
        <version>20.0</version>
    
    </dependency>
    
    <dependency>
    
        <groupId>org.apache.curator</groupId>
    
        <artifactId>curator-recipes</artifactId>
    
        <version>4.0.1</version>
    
    </dependency>
    
    <dependency>
    
        <groupId>org.apache.zookeeper</groupId>
    
        <artifactId>zookeeper</artifactId>
    
        <version>3.4.12</version>
    
    </dependency>

### 一、结构 ###

![70][]

### 二、逻辑 ###

1.采用guava做本地缓存服务

2.采用zookeeper做集群缓存通知

流程：

①修改、删除数据

![70 1][]

②读取数据（未命中缓存如下图，命中则直接返回）

![70 2][]

### 三、代码实现 ###

/***************本地缓存*******************/
    public class LocalCacheUtil{
    
        private static Map<String, Cache> cacheMap = new ConcurrentHashMap<>();
    
        private static ZkClient zkClient = new ZkClient("config/cacheZkConfig.properties");
    
        /**
         * 获取缓存内容
         * @param group 分组
         * @param key key
         * @return 缓存对象
         */
        public static Object get(String group, String key){
            PathUtil.check(group,"group");
            PathUtil.check(key,"key");
            Cache cache = cacheMap.get(group);
            if(cache == null){
                return null;
            }
            return cache.getIfPresent(key);
        }
    
        /**
         * 写缓存
         * @param group 分组
         * @param key key
         * @param val 值
         * @param expire 过期时间，不支持动态传入
         */
        public static void put(String group,String key,Object val,Long expire){
            PathUtil.check(group,"group");
            PathUtil.check(key,"key");
            Cache cache = getCache(group,expire);
            cache.put(key,val);
            zkClient.setWatcher(group, key, LocalCacheWatcher.getInstance());
        }
    
        /**
         * 删除缓存
         * @param group 分组
         * @param key key
         */
        public static void remove(String group,String key){
            PathUtil.check(group,"group");
            PathUtil.check(key,"key");
            Cache cache = cacheMap.get(group);
            if(cache != null){
                cache.invalidate(key);
            }
            zkClient.delete(group,key);
        }
    
        static void removeLocal(String group, String key){
            PathUtil.check(group,"group");
            PathUtil.check(key,"key");
            Cache cache = cacheMap.get(group);
            if(cache != null){
                cache.invalidate(key);
            }
        }
    
        static void removeAll(){
            for (String group : cacheMap.keySet()) {
                Cache cache = cacheMap.get(group);
                if(cache != null){
                    cache.invalidateAll();
                }
            }
        }
    
        /**
         * 获取cache
         * @param group 分组
         * @param expire 超时时间
         * @return cache
         */
        private static Cache getCache(String group, Long expire){
            Cache cache = cacheMap.get(group);
            if(cache == null){
                cache = CacheBuilder.newBuilder()
                        .maximumSize(2000)
                        .expireAfterWrite(expire, TimeUnit.MINUTES)
                        .build();
                cacheMap.put(group,cache);
            }
            return cache;
        }
    }
    
    
    /***************回调*******************/
    public class LocalCacheWatcher implements CuratorWatcher {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(LocalCacheWatcher.class);
    
        static LocalCacheWatcher getInstance(){
            return LocalCacheWatcherInstance.LOCAL_CACHE_WATCHER;
        }
    
        @Override
        public void process(WatchedEvent event) {
            LOGGER.info("收到监听事件："+event.toString());
            //空事件不处理，只做监控
            if(Watcher.Event.EventType.None.equals(event.getType())){
                if(Watcher.Event.KeeperState.Expired.equals(event.getState())){
                    LOGGER.info("清理全部缓存...");
                    LocalCacheUtil.removeAll();
                }
                return;
            }
            String path = event.getPath();
            String[] pathSplit = PathUtil.splitPath(path);
            LOGGER.info("GROUP = "+pathSplit[1]+",KEY = "+pathSplit[2]);
            LocalCacheUtil.removeLocal(pathSplit[1],pathSplit[2]);
        }
    
        private static class LocalCacheWatcherInstance{
            private static final LocalCacheWatcher LOCAL_CACHE_WATCHER = new LocalCacheWatcher();
        }
    }
    
    /***************zookeeper 处理类*******************/
    public class ZkClient {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(ZkClient.class);
        /**
         * zk客户端
         */
        private CuratorFramework cf;
    
        private String appPath;
    
        ZkClient(String path) {
            ZkConfig zkConfig = loadConfig(path);
            init(zkConfig);
        }
    
        public ZkClient(ZkConfig config) {
            init(config);
        }
    
        /**
         * zk初始化方法
         */
        private void init(ZkConfig zkConfig){
            LOGGER.info("ZkClientUtil init start...");
            cf = CuratorFrameworkFactory.builder()
                    .connectString(zkConfig.getConnectString())
                    .sessionTimeoutMs(zkConfig.getSessionTimeoutMs())
                    .connectionTimeoutMs(zkConfig.getConnectionTimeoutMs())
                    .retryPolicy(new ExponentialBackoffRetry(
                            zkConfig.getBaseSleepTimeMs(),
                            zkConfig.getMaxRetries()))
                    .namespace(zkConfig.getNamespace())
                    .build();
            cf.start();
            appPath = zkConfig.getAppPath();
            LOGGER.info("ZkClientUtil init complete...");
        }
    
        /**
         * 设置watcher
         * @param group 分组
         * @param key key
         * @param curatorWatcher watcher回调
         */
        void setWatcher(String group, String key, CuratorWatcher curatorWatcher)  {
    
            String path = PathUtil.getPath(appPath,group,key);
            try {
                Stat stat = cf.checkExists().forPath(path);
                if(stat == null){
                    cf.create().creatingParentsIfNeeded().forPath(path);
                }
                cf.checkExists().usingWatcher(curatorWatcher).forPath(path);
            } catch (Exception e) {
                LOGGER.error("set watcher fail, path:"+path+":"+e.getMessage());
            }
        }
    
        /**
         *  删除节点
         * @param group 分组
         * @param key key
         */
        void delete(String group, String key){
            String path = PathUtil.getPath(appPath,group,key);
            try {
                Stat stat = cf.checkExists().forPath(path);
                if(stat != null){
                    cf.delete().guaranteed().forPath(path);
                }
    
            } catch (Exception e) {
                LOGGER.error("delete zNode fail, path:"+path+":"+e.getMessage());
            }
        }
    
        private ZkConfig loadConfig(String path){
            LOGGER.info("ZkClient start load properties start :"+path);
            Properties properties = new Properties();
            // 使用ClassLoader加载properties配置文件生成对应的输入流
            InputStream in = ZkClient.class.getClassLoader().getResourceAsStream(path);
            // 使用properties对象加载输入流
            try {
                properties.load(in);
                LOGGER.info("ZkClient load properties finish :"+path);
            }catch (IOException e){
                throw new RuntimeException("ZkClient load properties fail :"+path,e);
            }
            ZkConfig zkConfig = new ZkConfig();
            zkConfig.setConnectString(properties.getProperty("connectString"));
            zkConfig.setSessionTimeoutMs(properties.getProperty("sessionTimeoutMs"));
            zkConfig.setConnectionTimeoutMs(properties.getProperty("connectionTimeoutMs"));
            zkConfig.setBaseSleepTimeMs(properties.getProperty("baseSleepTimeMs"));
            zkConfig.setMaxRetries(properties.getProperty("maxRetries"));
            zkConfig.setAppPath(properties.getProperty("appPath"));
            zkConfig.setNamespace(properties.getProperty("namespace"));
            return zkConfig;
        }
    }
    
    /*************ZkConfig zk配置类**************/
    public class ZkConfig {
        /**
         * 连接地址，多个地址用“,”间隔
         * 例如：localhost:2181,localhost1:2181
         */
        private String connectString;
        /**
         * session超时时间，单位ms
         */
        private int sessionTimeoutMs = 5000;
        /**
         * 连接超时时间，单位ms
         */
        private int connectionTimeoutMs = 5000;
        /**
         * 重试机制，间隔时间
         */
        private int baseSleepTimeMs = 100;
        /**
         * 重试机制，重试次数
         */
        private int maxRetries = 5;
        /**
         * 应用路径，用于隔离znood路径
         */
        private String appPath = "defaultAppPath";
        /**
         * namespace，用于区分zk用途
         */
        private String namespace;
    
        public String getConnectString() {
            return connectString;
        }
    
        public void setConnectString(String connectString) {
            if(connectString == null || "".equals(connectString.trim())){
                throw new RuntimeException("load properties fail，connectString must spacial!!!");
            }
            this.connectString = connectString;
        }
    
        public int getSessionTimeoutMs() {
            return sessionTimeoutMs;
        }
    
        public void setSessionTimeoutMs(String sessionTimeoutMs) {
            if(sessionTimeoutMs != null && !"".equals(sessionTimeoutMs.trim())){
                this.sessionTimeoutMs = Integer.valueOf(sessionTimeoutMs);
            }
        }
    
        public void setSessionTimeoutMs(int sessionTimeoutMs) {
            this.sessionTimeoutMs = sessionTimeoutMs;
        }
    
        public int getConnectionTimeoutMs() {
            return connectionTimeoutMs;
        }
    
        public void setConnectionTimeoutMs(String connectionTimeoutMs) {
            if(connectionTimeoutMs != null && !"".equals(connectionTimeoutMs.trim())){
                this.connectionTimeoutMs = Integer.valueOf(connectionTimeoutMs);
            }
        }
    
        public void setConnectionTimeoutMs(int connectionTimeoutMs) {
            this.connectionTimeoutMs = connectionTimeoutMs;
        }
    
        public int getBaseSleepTimeMs() {
            return baseSleepTimeMs;
        }
    
        public void setBaseSleepTimeMs(String baseSleepTimeMs) {
            if(baseSleepTimeMs != null && !"".equals(baseSleepTimeMs.trim())){
                this.baseSleepTimeMs = Integer.valueOf(baseSleepTimeMs);
            }
        }
    
        public void setBaseSleepTimeMs(int baseSleepTimeMs) {
            this.baseSleepTimeMs = baseSleepTimeMs;
        }
    
        public int getMaxRetries() {
            return maxRetries;
        }
    
        public void setMaxRetries(String maxRetries) {
            if(maxRetries != null && !"".equals(maxRetries.trim())){
                this.maxRetries = Integer.valueOf(maxRetries);
            }
        }
    
        public void setMaxRetries(int maxRetries) {
            this.maxRetries = maxRetries;
        }
    
        public String getAppPath() {
            return appPath;
        }
    
        public void setAppPath(String appPath) {
            if(appPath != null && !"".equals(appPath.trim())){
                this.appPath = appPath;
            }
        }
    
        public String getNamespace() {
            return namespace;
        }
    
        public void setNamespace(String namespace) {
            this.namespace = namespace;
        }
    }
    
    /*************PathUtil zk路径校验类*************/
    class PathUtil {
    
        private static final String PATH_SPLIT_CHAR = "/";
    
        private static final String PATH_SPLIT_CHAR1 = "\\";
    
        static String getPath(String... strs){
            StringBuilder stringBuilder = new StringBuilder();
            for (String str : strs) {
                stringBuilder.append(PATH_SPLIT_CHAR).append(str.trim());
            }
            return stringBuilder.toString();
        }
    
        static String[] splitPath(String path){
            path = path.substring(1);
            return path.split(PATH_SPLIT_CHAR);
        }
    
        static void check(String src, String name){
            if(src == null || "".equals(src.trim())){
                throw new RuntimeException(name +" not empty!");
            }
            //判断“/”,"\"，因为zk是路径
            if(src.contains(PATH_SPLIT_CHAR) || src.contains(PATH_SPLIT_CHAR1)){
                throw new RuntimeException(name +" can not contains \"/\" and \"\\\"!");
            }
        }
    }

### 四、使用 ###

public class LocalCacheTest {
        //模拟数据库
        private static final Map<String,Map<String,String>> db = new ConcurrentHashMap<>();
        //模拟数据库数据
        static{
            Map<String,String> map = new HashMap();
            map.put("uId","1001");
            map.put("name","name_"+"1001");
            db.put("1001",map);
        }
    
    
        public static void main(String[] args) throws InterruptedException {
            //设置配置
            ZkConfig cacheZkConfig = new ZkConfig();
            cacheZkConfig.setConnectString("localhost:2181");
            //初始化zk
            ZkClientUtil.init(cacheZkConfig);
    
            //1.测试读取缓存
            for (int i = 0 ;i<=1 ;i++) {
                System.out.println(findUser("1001"));
            }
            //2.测试缓存失效
    //        Thread.sleep(60000);
            System.out.println("--------------");
            System.out.println("过期后查询："+findUser("1001"));
            //3.测试缓存移除
            updateUser("1001","name_2");
            System.out.println("更新后查询："+findUser("1001"));
            //4.测试缓存移除
            deleteUser("1001");
            System.out.println("删除后查询："+findUser("1001"));
        }
    
        /**
         * 读取缓存测试
         * @param uId
         * @return
         */
        private static Map<String,String> findUser(String uId){
            String group = "user";
            Object o = LocalCacheUtil.get(group,uId);
            if(o != null){
                System.out.println("------->缓存命中，key："+uId);
                return (Map<String,String>) o;
            }
            System.out.println("------->缓存未命中，key："+uId);
            Map<String,String> user = db.get(uId);
            if(user != null){
                System.out.println("------->放缓存，key："+uId);
                LocalCacheUtil.put(group,uId,user,1L);
            }
            return user;
        }
    
        private static void updateUser(String uId,String name){
            String group = "user";
            Map<String,String> user = findUser(uId);
            if(user == null){
                throw new RuntimeException("user not exist,uId:"+uId);
            }
            System.out.println("------->删除缓存，key："+uId);
            LocalCacheUtil.remove(group,uId);
            user.put("name",name);
            db.put(uId,user);
            //如果担心此期间其他请求刷新缓存，可以在db修改后再remove一次缓存（缓存双淘汰）
        }
    
        private static void deleteUser(String uId){
            String group = "user";
            Map<String,String> user = findUser(uId);
            if(user == null){
                //不存在直接认为成功
                return;
            }
            System.out.println("------->删除缓存，key："+uId);
            LocalCacheUtil.remove(group,uId);
            db.remove(uId);
            //如果担心此期间其他请求刷新缓存，可以在db删除后再remove一次缓存（缓存双淘汰）
        }
    
    }

### 五、问题 ###

1.集群同步机制依赖于zookeeper，zk的建立连接是异步的，连不上也能启动，会影响集群通知

2.基于aop注解的没写

### 六、其他 ###

一次简单的尝试，欢迎讨（来）论（喷），邮箱地址laoxilaoxi@foxmail.com

[70]: /images/20220509/dcc91a4237394668b8354dce9d299835.png
[70 1]: /images/20220509/d888dcb07456474883b1d3a4bcddfe2f.png
[70 2]: /images/20220509/a396e244327849df9d42eb1330cbd8c5.png