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

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.2</version>
    </dependency>
</dependencies>

Application configuration (application.yml)

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

Java configuration

@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;
                }
            }
        }
    }
}

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!