Zookeeper + Guava loading cache 实现分布式缓存-白红宇

Zookeeper + Guava loading cache 实现分布式缓存

阅读量：6909 次

发布时间：2019-06-27

本文共 8254 字，大约阅读时间需要 27 分钟。

1. 概述

项目中，创建的活动内容存入redis，然后需要用到活动内容的地方，从redis去取，然后参与计算。

活动数据的一个特点是更新不频繁、数据量不大。因为项目部署一般是多机器、多实例，除了redis，有没有其他实现呢？

Guava的 loading cache是本地缓存，数据量不是很大时可以适用(如果有大量的key-value数据缓存本地，本机也吃不消啊)，

然后多机器多实例怎么同步呢？想到了zookeeper....

2. 代码

2.1 模拟多实例

package TestZK;import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.cache.PathChildrenCache;import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;import org.apache.curator.retry.ExponentialBackoffRetry;/** * @Author: rocky * @Date: Created in 2018/5/20. */public class InstanceOne {    private static final String ADDRESS = "xxx:2181";    private static final String PATH = "/strategy";    private static CuratorFramework client;    static {        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        client = CuratorFrameworkFactory.builder()                .connectString(ADDRESS)                .sessionTimeoutMs(5000)                .connectionTimeoutMs(5000)                .retryPolicy(retryPolicy)//                .namespace(BASE)                .build();        client.start();    }    public static void main(String[] args) throws Exception {        startCache();        Thread.sleep(Integer.MAX_VALUE);    }    private static void startCache() throws Exception {        PathChildrenCache childrenCache = new PathChildrenCache(client, PATH, true);        childrenCache.start();        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {                System.out.println("catch that: the path of changed node "+ event.getData().getPath()                        + ", the data of changed node is " + new String(event.getData().getData()));                //load data to loading cache(guava)                doSomething();            }        });    }    //load data to loading cache    private static void doSomething() {//        LoadingCache
      
       > driverIntegralCache = CacheBuilder.newBuilder().maximumSize(500)//                .expireAfterWrite(5, TimeUnit.MINUTES).build(new CacheLoader
       
        >() {//                    @Override//                    public Map
        
          load(String key) throws Exception {//                        Map
         
           integerDriverIntegralMap = ..;//                        logger.info("guava load ::driverIntegralMap"+integerDriverIntegralMap!=null?integerDriverIntegralMap.toString():"");//                        return integerDriverIntegralMap;//                    }//                });////        Map
          
            driverIntegralMap = driverIntegralCache.get(INTEGRAL);    }}package TestZK;import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.cache.PathChildrenCache;import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;import org.apache.curator.retry.ExponentialBackoffRetry;/** * @Author: rocky * @Date: Created in 2018/5/20. */public class InstanceTwo {    private static final String ADDRESS = "xxx:2181";    private static final String PATH = "/strategy";    private static CuratorFramework client;    static {        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        client = CuratorFrameworkFactory.builder()                .connectString(ADDRESS)                .sessionTimeoutMs(5000)                .connectionTimeoutMs(5000)                .retryPolicy(retryPolicy)//                .namespace(BASE)                .build();        client.start();    }    public static void main(String[] args) throws Exception {        startCache();        Thread.sleep(Integer.MAX_VALUE);    }    private static void startCache() throws Exception {        PathChildrenCache childrenCache = new PathChildrenCache(client, PATH, true);        childrenCache.start();        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {                System.out.println("catch that: the path of changed node "+ event.getData().getPath()                        + ", the data of changed node is " + new String(event.getData().getData()));            }        });    }}package TestZK;        import org.apache.curator.RetryPolicy;        import org.apache.curator.framework.CuratorFramework;        import org.apache.curator.framework.CuratorFrameworkFactory;        import org.apache.curator.framework.recipes.cache.PathChildrenCache;        import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;        import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;        import org.apache.curator.retry.ExponentialBackoffRetry;/** * @Author: rocky * @Date: Created in 2018/5/20. */public class InstanceThree {    private static final String ADDRESS = "xxx:2181";    private static final String PATH = "/strategy";    private static CuratorFramework client;    static {        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        client = CuratorFrameworkFactory.builder()                .connectString(ADDRESS)                .sessionTimeoutMs(5000)                .connectionTimeoutMs(5000)                .retryPolicy(retryPolicy)//                .namespace(BASE)                .build();        client.start();    }    public static void main(String[] args) throws Exception {        startCache();        Thread.sleep(Integer.MAX_VALUE);    }    private static void startCache() throws Exception {        PathChildrenCache childrenCache = new PathChildrenCache(client, PATH, true);        childrenCache.start();        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {                System.out.println("catch that: the path of changed node is "+ event.getData().getPath()                        + ", the data of changed node is " +  new String(event.getData().getData()));            }        });    }}

View Code

监听zk孩子节点，有变化时(创建、更新、删除)，重新从DB加载活动数据，缓存到loading cache，用到活动数据的地方，从cache中取。

2.2 模拟客户端

package TestZK;import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.retry.ExponentialBackoffRetry;/** * @Author: rocky * @Date: Created in 2018/5/20. */public class NodeChangeTest {    private static final String ADDRESS = "xxx:2181";    private static final String PATH = "/strategy";    private static CuratorFramework client;    static {        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        client = CuratorFrameworkFactory.builder()                .connectString(ADDRESS)                .sessionTimeoutMs(5000)                .connectionTimeoutMs(5000)                .retryPolicy(retryPolicy)//                .namespace(BASE)                .build();        client.start();    }    public static void main(String[] args) throws Exception {        String path_node_1 = PATH + "/node_1";        //create        client.create().creatingParentsIfNeeded().forPath(path_node_1);        Thread.sleep(2000);        //change        client.setData().forPath(path_node_1, "yahahaha".getBytes());        Thread.sleep(2000);        //delete        client.delete().forPath(path_node_1);        Thread.sleep(Integer.MAX_VALUE);    }}

View Code

客户端实际上可能是某个后台，在后台创建、修改、或删除某活动，然后对应操作zookeeper节点，其他监听到节点变化的地方，做相应操作(如查询DB并缓存数据)

控制台

3. 说明

Zookeeper原生的API只能实现一次监听，这里用到的Curator的封装jar包，免去了手动重复注册。另外Curator的NodeCache和TreeCache分别监听本节点及分支所有节点，

该实例演示的PathChildrenCache只能监控一级子节点(即儿子节点、孙子节点也不能监控)，根据需要选择相应NodeCache.

很low的实现，请多指教^_^

转载于:https://www.cnblogs.com/rocky-fang/p/9063968.html

你可能感兴趣的文章