How to Share Data Across Multiple JVMs in a Distributed Environment Using Zookeeper

In a distributed environment, Zookeeper can be used to share data among multiple JVM processes. The example uses Zookeeper 3.6.2 together with Spring Boot 2.2.10.

Required Maven dependencies

<code>&lt;dependencies&gt;
    &lt;dependency&gt;
        &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
        &lt;artifactId&gt;spring-boot-starter-web&lt;/artifactId&gt;
    &lt;/dependency&gt;
    &lt;dependency&gt;
        &lt;groupId&gt;org.apache.zookeeper&lt;/groupId&gt;
        &lt;artifactId&gt;zookeeper&lt;/artifactId&gt;
        &lt;version&gt;3.6.2&lt;/version&gt;
    &lt;/dependency&gt;
&lt;/dependencies&gt;</code>

Application configuration (application.yml)

<code>server:
  port: 9881
---
zk:
  connectString: localhost:2181,localhost:2182,localhost:2183
  timeout: 3000
  config:
    path: /config/ids
</code>

Java configuration

<code>@Configuration
public class ZKConfig {
    private static Logger logger = LoggerFactory.getLogger(ZKConfig.class);

    @Value("${zk.connectString}")
    private String connectString;

    @Value("${zk.timeout}")
    private Integer timeout;

    @Bean
    public ZooKeeper zookeeper() {
        ZooKeeper zookeeper = null;
        try {
            CountDownLatch cdl = new CountDownLatch(1);
            logger.info("zk准备连接: {}", connectString);
            zookeeper = new ZooKeeper(connectString, timeout, event -> {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    logger.info("zk服务器连接成功");
                    cdl.countDown();
                }
            });
            cdl.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return zookeeper;
    }

    @Configuration
    public static class WatchNodeData implements InitializingBean {
        public static Set<String> cache = new HashSet<>();

        @Resource
        private ZooKeeper zk;

        @Value("${zk.config.path}")
        private String path;

        @Override
        public void afterPropertiesSet() throws Exception {
            if (zk.exists(path, null) == null) {
                zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            // PERSISTENT_RECURSIVE watches child nodes recursively
            zk.addWatch(path, new NodeIdWatch(zk), AddWatchMode.PERSISTENT_RECURSIVE);
        }

        private static class NodeIdWatch implements Watcher {
            private ZooKeeper zk;
            public NodeIdWatch(ZooKeeper zk) { this.zk = zk; }
            @Override
            public void process(WatchedEvent event) {
                switch (event.getType()) {
                    case NodeDataChanged:
                        try {
                            String val = new String(this.zk.getData(event.getPath(), false, null));
                            logger.info("节点{} 数据发生变化, 当前值：{}", event.getPath(), val);
                            WatchNodeData.cache.add(val);
                        } catch (Exception e) { e.printStackTrace(); }
                        break;
                    case NodeCreated:
                        logger.info("创建了节点：{}", event.getPath());
                        break;
                    case NodeChildrenChanged:
                        logger.info("子节点发生了变化：{}", event.getPath());
                        break;
                    default:
                        break;
                }
            }
        }
    }
}
</code>

With the above configuration, the application watches for node data changes; whenever a change occurs, the new value is added to a

Set<String>

. This approach works for small data volumes but is not suitable for massive data sets that cannot be cached in JVM memory.

To update a value in Zookeeper, simply call

zk.setData

. Once the data is set, all services in the cluster receive the change through the watch mechanism.

Finished!