Nodejs 使用 ZooKeeper 做服务发现

将单体服务拆分为微服务后，为了服务高可用，一般会做集群多实例。但在分布式下，怎么进行高效、便捷的进行服务访问问题，出现了各类服务注册和服务发现框架。这里使用的是Zookeeper。ZooKeeper 官网 https://zookeeper.apache.org。

我们的业务系统使用的开发语言是JAVA，但是部分页面请求是先到nodejs 做的webportal服务，进行权限校验，校验通过后调用Java提供的API。当前阶段Java端已经微服务化，使用Zookeeper作为注册中心，目前只需要让nodejs端，也接入到Zookeeper，作为服务消费者，就能搭建机器环境。

找轮子

通过查找，发现npm有现成的库 node-zookeeper-client ,避免重复造轮子，就用它了。

接入思路

由于我们只是作为服务消费者，不需要使用服务注册的api，大部分可以直接在文档中找到API。

编码过程 

npm 安装

 npm i node-zookeeper-client 

连接ZK

const Zookeeper = require('node-zookeeper-client');
const CONNECTION_STRING = "127.0.0.1:2181"; // ZK的服务地址
const OPTIONS = {
  sessionTimeout: 5000 
}
const zk = Zookeeper.createClient(CONNECTION_STRING, OPTIONS);
  zk.on('connected', function(){
  console.log("zk=====", zk);
});
//获取根节点下的子节点数据
zk.getChildren('/', function(error, children, stat){
  if(error){
    console.log(error.stack);
    return;
  }
  console.log(children);
})
zk.connect();

其他API（仅供参考）

// 判断节点是否已存在
zk.exists('/phpnode',function(error,stat){
   if(stat){
        console.log("节点存在");
   }else{
       console.log("节点不存在");
   }
})
 
 // 创建/注册节点
zk.create('/phpnode',new Buffer('hello'),function(error,path){
   console.log(path);
})

// 获取节点数据
zk.getData('/phpnode',function(error,data,stat){
   console.log(data.toString());
});

//节点删除
zk.remove('/phpnode',function(error){
   if(!error){
       console.log('node 节点删除成功');
    }
})

于是有了第一版本代码

const zookeeper = require('node-zookeeper-client');


// ZK基础配置信息，正式项目需要从环境文件导入
export const ZK = {
    clientAddress: 'localhost:2181/zk/test', // ZK地址
    servicePath: '/test-service', // 服务路径
};

let zkClient = null;

// 获取服务ip+port
export const getZKServiceBaseUrl = (servicePath) => {
    return new Promise((resolve, reject) => {
        try {
            // 防止重复连接
            if (zkClient) {
                disconnectZKService();
            }

            // 新建连接
            zkClient = zookeeper.createClient(ZK.clientAddress);
            // 连接后执行一次
            zkClient.once('connected', async function () {
                // 获取服务节点信息
                const res = await listChildren(zkClient, servicePath);
                res.message ? reject(res) : resolve(res);
            });

            zkClient.connect();
        } catch (error) {
            reject(error);
        }
    });
};

// 断开链接
export const disconnectZKService = () => {
    if (zkClient) {
        zkClient.close();
    }
};

// 获取节点信息，ip+port
function listChildren(client, path) {
    return new Promise((resolve, reject) => {
        client.getChildren(path,
            function () {},
            function (error, children) {
                if (error) {
                    reject({
                        ...error,
                        message: `获取ZK节点error，Path: ${path}`
                    });
                }
                try {
                    let addressPath = path + '/';
                    if (children.length > 1) {
                        //若存在多个地址，则随机获取一个地址
                        addressPath += children[Math.floor(Math.random() * children.length)];
                    } else {
                        //若只有唯一地址，则获取该地址
                        addressPath += children[0];
                    }
                    //获取服务地址
                    client.getData(addressPath, function (err, data) {
                        if (err) {
                            reject({
                                ...error,
                                message: `获取ZK服务地址error，Stack: ${err.stack}`
                            });
                        }
                        if (!data) {
                            reject({
                                ...error,
                                message: `ZK data is not exist`
                            });
                        }
                        const serviceInfo = JSON.parse(data);

                        const url = serviceInfo.address + ':' + serviceInfo.port;
                        resolve(url);
                    });
                } catch (error) {
                    reject({
                        ...error,
                        message: `list ZK children error`
                    });
                }
            }
        );
    });
}

通过测试代码，可以实现调用Java服务。可能一般的程序员实现功能了就好了，可是作为一个有点追求的，感觉代码哪里有问题。具体是哪里呢，盯着屏幕瞅了两分钟，发现每次获取服务都取 ZK 注册中心获取，这个过程涉及到的网络请求而且还不是一次HTTP，如果只是这么简单的改造，程序单纯在性能响应上很有可能还不如老版本。我们可以在获取服务的真实远程地址前，添加一个本地缓存。通过ZK订阅机制，更新本地缓存数据。

思路虽然明确了，可以api扫了扫，没有我们想要的监听器，如下所示

 

这怎么办，按理说的应该会有一个，节点数据改变推送的监听器，例如新增，删除，修改等等。找了半天也没找到合适的。

没办法，接着看源码吧，看了一会，忽然，看到一个似乎可用的，类

 

 这不就是我需要的类吗，但是居然在一方法中注入监听器，先试试吧。

 

 试了一下，嘿，真的可以了，当服务端节点数据发生变动后，会自动触发监听器 watcher 的回调逻辑。这就好办了，改造开始。

改进后的代码

const zookeeper = require('node-zookeeper-client');
var ZK = require('../config/env.js').zk;

const client = Object.freeze({
    zkClient: zookeeper.createClient(ZK.connectionString),
    serviceSet: [], //
    serviceCache: Object.freeze({
        map: new Map(),

        /**
         * 更新缓存
         * @param {String} path 服务路径
         * @param {Array} arr 真实访问集合
         */
        updateCache: function (path, arr) {
            this.map.set(path, arr);
        },

        /**
         * 从缓存中获取访问地址
         * 
         * @param {String} path 服务路径
         * @returns String 真实访问地址
         */
        getRealPath: function (path) {
            let arr = this.map.get(path);

            if (arr.length > 1) //若存在多个地址，则随机获取一个地址
                return arr[Math.floor(Math.random() * arr.length)];
            else //若只有唯一地址，则获取该地址
                return arr[0];
        }
    }),

    connect: function () {
        console.info("连接 zookeeper");

        this.zkClient.once('connected', function () {
            console.info("连接成功");
        });

        this.zkClient.connect();
    },

    getRealPath: function (serviceName) {
        return new Promise(async (resolve, reject) => {
            if (this.serviceSet.includes(serviceName)) {
                resolve(this.serviceCache.getRealPath(serviceName));
            } else {
                // 加载服务节点信息
                this.loadChildren(serviceName).then(url => resolve(url)).catch(error => reject(error));
            }
        });
    },

    loadChildren: function (path) {
        console.info("进入 loadChildren ");
        return new Promise((resolve, reject) => {
            this.zkClient.getChildren(path, (event) => {
                console.info(" loadChildren watcher ", path, event);

                this.getChildren(event.path);
            }, (error, ids) => {
                console.info(" loadChildren callback ", path, error, ids);
                if (error) {
                    reject({
                        ...error,
                        message: `获取ZK节点error，Path: ${path}`
                    });
                } else {
                    resolve(this.getData(path, ids));
                }
            });
        });
    },

    getChildren: function (path) {
        console.info("进入 getChildren ");
        return new Promise((resolve, reject) => {
            this.zkClient.getChildren(path, (error, ids) => {
                console.info(" getChildren callback ", path, error, ids);
                if (error) {
                    reject({
                        ...error,
                        message: `获取ZK节点error，Path: ${path}`
                    });
                }

                resolve(this.getData(path, ids));
            });
        });
    },

    getData: function (path, ids) {
        console.info("进入 getData ");

        let pros = ids.map(id => new Promise((resolve, reject) => {
            //获取服务地址
            this.zkClient.getData(path + "/" + id, (error, data) => {
                console.info(" getData callback ", path, id);
                if (error) {
                    reject({
                        ...error,
                        message: `获取ZK服务地址error，Stack: ${err.stack}`
                    });
                }
                if (!data) {
                    reject({
                        ...error,
                        message: `ZK data is not exist`
                    });
                }

                const node = JSON.parse(data).payload;
                const protocol = node.ssl ? "https://" : "http://";

                resolve(`${protocol}${node.host}:${node.port}`);
            });
        }));

        return Promise.all(pros).then(arr => this.serviceCache.updateCache(path, arr)).then(() => this.serviceCache.getRealPath(path));
    },

    disconnect: function () { //断开连接
        console.info("进入 disconnect ")
        if (this.zkClient) {
            console.info("执行 close")
            this.zkClient.close();
        }
    },
});

client.connect();

module.exports = {
    getServiceUrl: (path) => client.getRealPath(path),
    disconnect: () => client.disconnect(),
}

这样终于，好一点了。

未完待续（多节点的选择问题）

多节点选择策略：随机，轮转，粘性 等等，一般不同的项目使用的策略也不太一样，实例中使用的是简单随机策略，后续再进行节点选择的策略问题优化啦。

关机，收工！！！