开发环境

参考

基于ZOOKEEPER+THRIFT的RPC动态服务注册发现和调用(JAVA)

如何安装windows版本的zookeeper

Thrift 环境搭建及 Java Demo 运行

源码下载

demo-thrift.zip

Windows 版本 ZooKeeper 安装

下载

去官网下载安装包 apache-zookeeper-3.8.0-bin.tar.gz

Apache 官网下载地址：https://zookeeper.apache.org/releases.html

解压

解压 apache-zookeeper-3.8.0-bin.tar.gz 到 D:\Program Files\apache-zookeeper-3.8.0-bin

新建目录

D:\Program Files\apache-zookeeper-3.8.0-bin\data
D:\Program Files\apache-zookeeper-3.8.0-bin\log

修改配置

复制 D:\Program Files\apache-zookeeper-3.8.0-bin\conf\zoo_sample.cfg
命名为 D:\Program Files\apache-zookeeper-3.8.0-bin\conf\zoo.cfg

打开 zoo.cfg，修改下面内容

dataDir=D:\Program Files\apache-zookeeper-3.8.0-bin\data
dataLogDir=D:\Program Files\apache-zookeeper-3.8.0-bin\log

启动 ZooKeeper 服务器端

运行

D:\Program Files\apache-zookeeper-3.8.0-bin\bin\zkServer.cmd

启动 ZooKeeper 客户端

运行

D:\Program Files\apache-zookeeper-3.8.0-bin\bin\zkCli.cmd

出现“Welcome to ZooKeeper!”，运行成功

IDEA Zookeeper 插件安装

安装插件 Zookeeper tool

侧边栏即可查看

项目结构

新建项目-demo-thrift

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>demo.thrift</groupId>
    <artifactId>demo-thrift</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>demo-thrift-contract</module>
        <module>demo-thrift-common</module>
        <module>demo-thrift-server</module>
        <module>demo-thrift-web</module>
    </modules>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-collections4</artifactId>
            <version>4.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-x-discovery</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-test</artifactId>
            <version>4.0.0</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

新建模块-demo-thrift-contract

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo-thrift</artifactId>
        <groupId>demo.thrift</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>demo-thrift-contract</artifactId>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.thrift</groupId>
            <artifactId>libthrift</artifactId>
            <version>0.9.2</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>${project.name}</finalName>
        <plugins>
            <plugin>
                <groupId>org.apache.thrift.tools</groupId>
                <artifactId>maven-thrift-plugin</artifactId>
                <version>0.1.11</version>
                <configuration>
                    <!-- 如果thrift执行文件的路径已经设置到PATH环境变量，可以直接简写。否则需要给出完整执行路径 -->
                    <thriftExecutable>thrift</thriftExecutable>
                    <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
                </configuration>
                <executions>
                    <execution>
                        <id>thrift-sources</id>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>thrift-test-sources</id>
                        <phase>generate-test-sources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

thrift

CommonResult.thrift

namespace java demo.thrift.contract
exception CommonException{
    1:i32 errorCode,
    2:string message
}
struct CommonResultDto{
    1:bool flag = true;             //执行标记
    2:string message;               //错误信息
    3:string data;                  //数据
    4:bool overflow = false;        //溢出标记
}

TestTableService.thrift

namespace java demo.thrift.contract
include "CommonResult.thrift"
struct TestTableDto{
    1:i32    id,
    2:string no,
    3:string name,
    4:i32    isDelete
}
struct TestTablePageDto{
    1:optional i32 total = 0
    2:optional list<TestTableDto> resultDtos
}
service TestTableService {
    CommonResult.CommonResultDto add(1:TestTableDto param) throws(1:CommonResult.CommonException ex)
}

生成 Java 代码

demo.thrift.contract 目录下自动生成 Java 代码

新建模块-demo-thrift-common

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo-thrift</artifactId>
        <groupId>demo.thrift</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>demo-thrift-common</artifactId>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
</project>

ZookeeperUtils

package demo.jarvis.common.Utils;
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.PathChildrenCacheListener;
import org.apache.curator.retry.RetryForever;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.ACL;
import java.util.List;
public class ZookeeperUtils implements AutoCloseable {
    /**
     * zk 服务名称
     */
    private String name;
    /**
     * zk 服务器连接字符串
     */
    private String zkConnectedStr;
    /**
     * 封装好的client
     */
    private CuratorFramework client;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getZkConnectedStr() {
        return zkConnectedStr;
    }
    public void setZkConnectedStr(String zkConnectedStr) {
        this.zkConnectedStr = zkConnectedStr;
    }
    public CuratorFramework getClient() {
        return client;
    }
    public void setClient(CuratorFramework client) {
        this.client = client;
    }
    /**
     * 自定义一个异常捕获处理器,只是打印暂时无别的操作
     */
    private Thread.UncaughtExceptionHandler uncaughtExceptionHandler = new Thread.UncaughtExceptionHandler() {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            System.out.println(name + ": " + e);
        }
    };
    /**
     * 构造函数，启动zk客户端和连接zk服务器
     *
     * @param name           zk 服务名称
     * @param zkConnectedStr zk 服务器连接字符串
     */
    public ZookeeperUtils(String name, String zkConnectedStr) throws Exception {
        try {
            this.name = name;
            this.zkConnectedStr = zkConnectedStr;
            RetryPolicy retryPolicy = new RetryForever(10000);
            this.client = CuratorFrameworkFactory.builder()
                    .connectString(zkConnectedStr)
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(30 * 1000)
                    .connectionTimeoutMs(30 * 1000)
                    .maxCloseWaitMs(60 * 1000)
                    .threadFactory(new ThreadFactoryBuilder().setNameFormat(name + "-%d").setUncaughtExceptionHandler(uncaughtExceptionHandler).build())
                    .build();
            this.client.start();
            this.client.blockUntilConnected();
            System.out.println(String.format("zk :  %s started.", this.zkConnectedStr));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new Exception(e);
        }
    }
    /**
     * 添加节点
     *
     * @param path
     * @param value
     * @param mode
     * @return
     */
    public String add(String path, byte[] value, CreateMode mode) throws Exception {
        try {
            return this.client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, value);
        } catch (Exception e) {
            throw new Exception(e);
        }
    }
    /**
     * 添加节点
     *
     * @param path
     * @param value
     * @param mode
     * @param aclList
     * @return
     */
    public String add(String path, byte[] value, CreateMode mode, List<ACL> aclList) throws Exception {
        try {
            return this.client.create().creatingParentsIfNeeded().withMode(mode).withACL(aclList).forPath(path, value);
        } catch (Exception e) {
            throw new Exception(e);
        }
    }
    /**
     * 判断节点是否存在
     *
     * @param path
     * @return
     */
    public boolean exist(String path) {
        try {
            return this.client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            System.out.println(String.format("zk check exist error, path = %s , %s", path, e));
            return false;
        }
    }
    /**
     * 移除节点
     *
     * @param path
     */
    public void remove(String path) throws Exception {
        try {
            this.client.delete().forPath(path);
        } catch (Exception e) {
            throw new Exception(e);
        }
    }
    /**
     * 设置节点数据
     *
     * @param path
     * @param value
     */
    public void set(String path, byte[] value) throws Exception{
        try {
            this.client.setData().forPath(path, value);
        } catch (Exception e) {
            throw new Exception(e);
        }
    }
    /**
     * 获取节点下的所有子节点
     *
     * @param nodePath
     * @return
     */
    public List<String> getChildren(String nodePath) {
        try {
            return this.client.getChildren().forPath(nodePath);
        } catch (Exception e) {
            System.out.println(String.format("get node children failed, nodePath = %s ", nodePath));
        }
        return null;
    }
    /**
     * 注册目录监听器
     *
     * @param nodePath
     * @param listener
     * @return
     */
    public PathChildrenCache registerPathChildrenListener(String nodePath, PathChildrenCacheListener listener) {
        try {
            //创建一个PathChildrenCache
            PathChildrenCache pathChildrenCache = new PathChildrenCache(this.client, nodePath, true);
            //添加子目录监视器
            pathChildrenCache.getListenable().addListener(listener);
            //启动监听器
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            //返回PathChildrenCache
            return pathChildrenCache;
        } catch (Exception e) {
            System.out.println(String.format("register path children node cache listener failed, nodePath = %s ", nodePath));
        }
        return null;
    }
    @Override
    public void close() throws Exception {
        System.out.println(String.format("zk %s - %s closed", this.name, this.zkConnectedStr));
        CloseableUtils.closeQuietly(this.client);
    }
}

新建模块-demo-thrift-server

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo-thrift</artifactId>
        <groupId>demo.thrift</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>demo-thrift-server</artifactId>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>demo.thrift</groupId>
            <artifactId>demo-thrift-contract</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>demo.thrift</groupId>
            <artifactId>demo-thrift-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>
</project>

TestTableServiceImpl

package com.thrift.server.serviceImpl;
import demo.thrift.contract.CommonResultDto;
import demo.thrift.contract.TestTableDto;
import demo.thrift.contract.TestTableService;
import org.apache.thrift.TException;
public class TestTableServiceImpl implements TestTableService.Iface {
    @Override
    public CommonResultDto add(TestTableDto param) throws TException{
        CommonResultDto resultDto = new CommonResultDto();
        resultDto.setData("insert:" + param.getNo()+"-"+param.getName());
        return resultDto;
    }
}

ThriftServer

package com.thrift.server;
import com.thrift.server.serviceImpl.TestTableServiceImpl;
import demo.jarvis.common.Utils.ZookeeperUtils;
import demo.thrift.contract.TestTableService;
import org.apache.commons.lang3.StringUtils;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.zookeeper.CreateMode;
import java.net.InetAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class ThriftServer{
    /**
     * RPC 所有相关节点公用的顶级父节点
     */
    private static final String THRIFT_SERVER_PREFIX = "/thrift/";
    /**
     * 固定的一个单线程池
     */
    private ExecutorService thread = Executors.newSingleThreadExecutor();
    /**
     * zk 客户端服务实例
     */
    private ZookeeperUtils zookeeperUtils;
    /**
     * RPC 服务名
     */
    private String name;
    /**
     * PRC 服务端口号
     */
    private int port;
    /**
     * 构造函数，启动RPC节点并注册节点到ZK
     *
     * @param port           PRC 服务端口号
     * @param name           RPC 服务名
     * @param zookeeperUtils zk 客户端服务实例
     */
    public ThriftServer(int port, String name, ZookeeperUtils zookeeperUtils) throws Exception{
        this.zookeeperUtils = zookeeperUtils;
        this.name = name;
        this.port = port;
        this.startAndRegisterService();
    }
    /**
     * 获取本服务即将注册到ZK上的节点路径
     * 根据本机器IP和定义的端口号生成唯一路径
     *
     * @return 路径
     */
    private String getServiceNodePath() {
        try {
            InetAddress inetAddress = InetAddress.getLocalHost();
            String servicePath = inetAddress.getHostAddress();
            return "/".concat(servicePath).concat(":").concat(String.valueOf(port));
        } catch (Exception e) {
            return null;
        }
    }
    /**
     * 注册 RPC 服务父节点
     * 比如我们此 RPC 服务为 hello-server
     * 注册完之后就是 /thrift/hello-server 永久节点
     */
    private void register() throws Exception{
        if (!zookeeperUtils.exist(THRIFT_SERVER_PREFIX.concat(name))) {
            zookeeperUtils.add(THRIFT_SERVER_PREFIX.concat(name), name.getBytes(), CreateMode.PERSISTENT);
        }
    }
    /**
     * 注册此 RPC 服务节点
     * 注册完之后就是 /thrift/hello-server/10.1.38.226:7778 临时节点
     */
    private void registerService() throws Exception {
        if (!zookeeperUtils.exist(THRIFT_SERVER_PREFIX.concat(name))) {
            register();
        }
        String serviceNodePath = getServiceNodePath();
        if (StringUtils.isBlank(serviceNodePath)) {
            return;
        }
        zookeeperUtils.add(THRIFT_SERVER_PREFIX.concat(name).concat(serviceNodePath), String.valueOf(port).getBytes(), CreateMode.EPHEMERAL);
    }
    /**
     * 启动 RPC 服务
     * @return 启动成功返回 true
     */
    private boolean start() {
        try {
            //构造 thrift-server
            TServer server = new TThreadedSelectorServer(new TThreadedSelectorServer.Args(new TNonblockingServerSocket(port))
                    .protocolFactory(new TBinaryProtocol.Factory())
                    .processor(new TestTableService.Processor(new TestTableServiceImpl()))
                    .workerThreads(5)
                    .transportFactory(new TFramedTransport.Factory()));
            //异步线程提交,防止主线程阻塞
            thread.submit(server::serve);
            //一直轮训等待RPC服务启动成功,服务正常
            while (!server.isServing()) {
                System.out.println("wait for thrift server start!");
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    System.out.println(e);
                }
            }
        } catch (TTransportException e) {
            System.out.println(e);
            return false;
        }
        return true;
    }
    /**
     * 启动 RPC 服务并且注册 ZK 节点
     */
    private void startAndRegisterService() throws Exception {
        if (!start()) {
            System.out.println(name.concat("start failed!"));
        }
        registerService();
    }
    public static void main(String[] args) throws Exception{
        //创建 ZK 客户端实例
        ZookeeperUtils zookeeperUtils = new ZookeeperUtils("thrift-test", "127.0.0.1:2181");
        //异步启动第一个 RPC 服务
        new Thread("thrift-server-1") {
            @Override
            public void run() {
                try {
                    new ThriftServer(7777, "thrift-server", zookeeperUtils);
                }catch (Exception e){
                    System.out.println(e.getMessage());
                }
            }
        }.start();
        //异步启动第二个 RPC 服务
        new Thread("thrift-server-2") {
            @Override
            public void run() {
                try {
                    new ThriftServer(7778, "thrift-server", zookeeperUtils);
                }catch (Exception e){
                    System.out.println(e.getMessage());
                }
            }
        }.start();
    }
}

新建模块-demo-thrift-web

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo-thrift</artifactId>
        <groupId>demo.thrift</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>demo-thrift-web</artifactId>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>demo.thrift</groupId>
            <artifactId>demo-thrift-contract</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>demo.thrift</groupId>
            <artifactId>demo-thrift-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>
</project>

ThriftWeb

package demo.thrift.web;
import demo.jarvis.common.Utils.ZookeeperUtils;
import demo.thrift.contract.CommonResultDto;
import demo.thrift.contract.TestTableDto;
import demo.thrift.contract.TestTableService;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
public class ThriftWeb {
    /**
     * 自己知道自己的 RPC 服务节点路径所以写死
     */
    private static final String NODE_PATH = "/thrift/thrift-server";
    /**
     * IP 和端口号分隔符
     */
    private static final String SPLIT_STR = ":";
    /**
     * ZK 客户端实例
     */
    private ZookeeperUtils zookeeperUtils;
    /**
     * RPC 服务连接池,Map 本地简单存储
     */
    private Map<String, TProtocol> protocolMap = new ConcurrentHashMap<>();
    /**
     * 构造函数，RPC 服务发现和 RPC 服务动态监听
     * @param zookeeperUtils ZK 客户端实例
     */
    public ThriftWeb(ZookeeperUtils zookeeperUtils) {
        this.zookeeperUtils = zookeeperUtils;
        this.serverDetect();
        this.serverListening();
    }
    /**
     * 获取一个 RPC 服务连接
     * @return RPC 服务连接
     */
    private TProtocol getProtocol() {
        //如果连接池为空，则返回空
        if (MapUtils.isEmpty(protocolMap)) {
            return null;
        }
        //随机取出一个RPC服务连接
        List<TProtocol> tmp = new ArrayList<>(new ArrayList<>(protocolMap.values()));
        Collections.shuffle(tmp);
        //打印一下使用的那个连接
        System.out.println(tmp.get(0).toString());
        return tmp.get(0);
    }
    /**
     * 对应 zk 节点路径的 RPC 服务进连接池
     * @param path RPC 服务节点路径
     */
    private void inProtocolPool(String path) {
        if (StringUtils.isBlank(path)) {
            return;
        }
        //解析该节点的服务地址
        String[] address = path.split(SPLIT_STR);
        if (ArrayUtils.isEmpty(address) ||  address.length != 2) {
            return;
        }
        if (protocolMap.containsKey(path)) {
            return;
        }
        try {
            //创建与 RPC 服务端的连接
            TTransport tTransport = new TSocket(address[0], Integer.parseInt(address[1]));
            TFramedTransport framedTransport = new TFramedTransport(tTransport);
            TProtocol tProtocol = new TBinaryProtocol(framedTransport);
            tTransport.open();
            //进连接池
            protocolMap.put(path, tProtocol);
        } catch (TTransportException e) {
            System.out.println(e);
        }
    }
    /**
     * 对应 zk 节点路径的 RPC 服务出连接池
     * @param path RPC 服务节点路径
     */
    private void outProtocolPool(String path) {
        if (!protocolMap.containsKey(path)) {
            return;
        }
        //出连接池
        TProtocol protocol = protocolMap.remove(path);
        //关闭连接
        protocol.getTransport().close();
    }
    /**
     * 发现 RPC 服务
     */
    private void serverDetect() {
        List<String> childrenNodePaths = zookeeperUtils.getChildren(NODE_PATH);
        if (childrenNodePaths != null) {
            childrenNodePaths.forEach(this::inProtocolPool);
        }
    }
    /**
     * 动态监听 RPC 服务
     */
    private void serverListening() {
        //注册 ZK 目录监听器
        zookeeperUtils.registerPathChildrenListener(NODE_PATH, (curatorClient, event) -> {
            //获取变化的节点数据
            ChildData childData = event.getData();
            if (childData == null) {
                return;
            }
            switch (event.getType()) {
                case CHILD_ADDED://新增RPC节点
                    System.out.println(String.format("path children add children node %s now", childData.getPath()));
                    //新节点进RPC服务连接池
                    inProtocolPool(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                case CHILD_REMOVED://减少RPC节点
                    System.out.println(String.format("path children delete children node %s now", childData.getPath()));
                    //失去的节点出RPC服务连接池
                    outProtocolPool(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                case CONNECTION_LOST://RPC节点连接丢失
                    System.out.println(String.format("path children connection lost %s now", childData.getPath()));
                    //断开连接节点出RPC服务连接池
                    outProtocolPool(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                case CONNECTION_RECONNECTED://RPC节点重连
                    System.out.println(String.format("path children connection reconnected %s now", childData.getPath()));
                    //重新连接的节点出RPC服务连接池
                    inProtocolPool(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                default://无操作
                    break;
            }
        });
    }
    /**
     * 客户端 addTestTable
     *
     * @param id
     * @return
     */
    private CommonResultDto addTestTable(TestTableDto param) {
        //获取一个 RPC 服务连接
        TProtocol protocol = this.getProtocol();
        if (protocol == null) {
            return null;
        }
        //创建一个RPC实例
        TestTableService.Client client = new TestTableService.Client(protocol);
        try {
            return client.add(param);
        } catch (TException e) {
            System.out.println(e);
        }
        return null;
    }
    public static void main(String[] args) throws Exception {
        //ZK 客户端实例
        ZookeeperUtils zookeeperUtils = new ZookeeperUtils("thrift-test", "127.0.0.1:2181");
        //Thrift 客户端
        ThriftWeb thriftWeb = new ThriftWeb(zookeeperUtils);
        //每五秒打印三次 addTestTable 的结果,结果可想而知，使用的rpc服务不是同一个，会随机选取调用
        while (true) {
            for (int i = 0; i < 3; i++) {
                TestTableDto testTableDto = new TestTableDto();
                testTableDto.setNo("no-" + i);
                testTableDto.setName("name-" + i);
                System.out.println(thriftWeb.addTestTable(testTableDto).getData());
            }
            TimeUnit.SECONDS.sleep(5);
        }
    }
}

测试

demo-thrift-server

● 启动 demo-thrift-server - ThriftServer - main()

控制台输出

log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
zk :  127.0.0.1:2181 started.
wait for thrift server start!
wait for thrift server start!

demo-thrift-web

● 启动 demo-thrift-web - ThriftWeb - main()

控制台输出

log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
zk :  127.0.0.1:2181 started.
org.apache.thrift.protocol.TBinaryProtocol@536aaa8d
insert:no-0-name-0
org.apache.thrift.protocol.TBinaryProtocol@536aaa8d
insert:no-1-name-1
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-2-name-2
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-0-name-0
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-1-name-1
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-2-name-2
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-0-name-0
org.apache.thrift.protocol.TBinaryProtocol@528931cf
insert:no-1-name-1
org.apache.thrift.protocol.TBinaryProtocol@536aaa8d
insert:no-2-name-2