基于Zookeeper注册中心实现简易手写RPC框架

r囧r小猫 2022-05-16 02:59 243阅读 0赞

# 基于Zookeeper注册中心实现简易手写RPC框架 #

## 一，环境准备 ##

1.  Zookepper单点或集群环境，演示使用集群环境
2.  Zookeeper的Curator客户端工具
3.  使用Maven进行项目构建

## 二，代码实现 ##

### 服务端开启服务 ###

1，jar包依赖

<dependencies>
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.11</version>
          <scope>test</scope>
        </dependency>
        <dependency>
          <groupId>org.apache.zookeeper</groupId>
          <artifactId>zookeeper</artifactId>
          <version>3.4.8</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-client</artifactId>
          <version>4.0.0</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.curator</groupId>
          <artifactId>curator-recipes</artifactId>
          <version>4.0.0</version>
        </dependency>
      </dependencies>

2，VO和服务地址

\* VO

public class RPCRequest implements Serializable {
        private static final long serialVersionUID = -406351179958827029L;
    
        private String className;
    
        private String methodName;
    
        private Object[] parameters;
    
        private String version;
    
        public String getVersion() {
            return version;
        }
    
        public void setVersion(String version) {
            this.version = version;
        }
    
        public String getClassName() {
            return className;
        }
    
        public void setClassName(String className) {
            this.className = className;
        }
    
        public String getMethodName() {
            return methodName;
        }
    
        public void setMethodName(String methodName) {
            this.methodName = methodName;
        }
    
        public Object[] getParameters() {
            return parameters;
        }
    
        public void setParameters(Object[] parameters) {
            this.parameters = parameters;
        }
    }

\* 服务地址

public class ZKConfig {
    
        public static final String ZK_HOST = "192.168.91.128:2181,192.168.91.129:2181,192.168.91.130:2181";
    
        public static final String ZK_REGISTER_PATH = "/registers";
    }

\* 自定义注解

@Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    public @interface RpcAnnotation {
    
        /**
         * 对外发布的服务接口地址
         */
        Class<?> value();
    
        /**
         * 版本号
         */
        String version() default "";
    
    }

3，注册服务接口

\* 通过构造方法初始化Curator

private CuratorFramework curatorFramework;
    
        public RegisterService() {
            curatorFramework = CuratorFrameworkFactory.builder()
                    .connectString(ZKConfig.ZK_HOST).sessionTimeoutMs(4000)
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
            curatorFramework.start();
        }

\* 注册服务

-- 注册服务名称为根节点下的持久化节点

-- 注册服务对应的服务器地址为服务名称节点下的临时节点

-- 每次服务重启时会清空临时节点（服务的分布式服务器），并重新注册临时节点

public void register(String serviceName, String serviceAddress) {
            try {
                // 注册相应的服务 --> 不存在则创建
                String serviceUrl = ZKConfig.ZK_REGISTER_PATH + "/" + serviceName;
                if (null == curatorFramework.checkExists().forPath(serviceUrl)) {
                    curatorFramework.create().creatingParentsIfNeeded()
                            .withMode(CreateMode.PERSISTENT).forPath(serviceUrl, "0".getBytes());
                }
                // 获取地址的完整路径
                String addressUrl = serviceUrl + "/" + serviceAddress;
                // 注册地址的临时节点
                String rsNode = curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(addressUrl, "0".getBytes());
                System.out.println("服务注册成功, rsNode : " + rsNode);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

4，绑定服务

\* 通过构造初始化构造中心，服务发布地址已经服务绑定集合（Map）

public class RPCServer {
    
        private final ExecutorService executorService = Executors.newCachedThreadPool();
    
        /**
         * 注册中心
         */
        private IRegisterService registerService;
    
        /**
         * 服务发布地址
         */
        private String serviceAddress;
    
        Map<String, Object> handlerMap = new HashMap<String, Object>();
    
        public RPCServer(IRegisterService registerService, String serviceAddress) {
            this.registerService = registerService;
            this.serviceAddress = serviceAddress;
        }

\* 对外服务接口

@RpcAnnotation(value = ISelfHelloService.class)
    public class SelfHelloService implements ISelfHelloService {
    
        @Override
        public String sayHello(String msg) {
            return "[I'M 8081]HELLO : " + msg;
        }
    }

\* 根据注解信息获取服务名称，服务版本号并绑定服务

public void bind(Object ... services) {
            for (Object service : services) {
                RpcAnnotation annotation = service.getClass().getAnnotation(RpcAnnotation.class);
                String serviceName = annotation.value().getName();
                String version = annotation.version();
                if (null != version && !"".equals(version)) {
                    serviceName = serviceName + "-" + version;
                }
                // 绑定服务接口名称对应的服务
                handlerMap.put(serviceName, service);
            }
        }

5，发布服务端服务

\* 根据服务地址启动服务，并注册服务到注册中心

public void publishServer() {
            ServerSocket serverSocket = null;
            try {
                // 服务端启动服务代码内容==================
                // 启动服务
                serverSocket = new ServerSocket(Integer.valueOf(serviceAddress.split(":", -1)[1]));
                for (String interfaceName : handlerMap.keySet()) {
                    //
                    registerService.register(interfaceName, serviceAddress);
                    System.out.println("注册服务成功, interfaceName : " + interfaceName + ", serviceAddress : " + serviceAddress);
                }
                
                // 服务端获取客户端请求代码处理======================
                // 获取监听
                while (true) {
                    Socket socket = serverSocket.accept();
                    executorService.execute(new ProcessorHandler(socket, handlerMap));
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (null != serverSocket) {
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }

\* main方法启动服务，通过启动两个服务实现模拟分布式，并验证负载均衡

public class ClusterServerService_1 {
        public static void main(String[] args) {
            ISelfHelloService helloService = new SelfHelloService();
            IRegisterService registerService = new RegisterService();
            // 初始化注册中心和服务端口信息
            RPCServer server = new RPCServer(registerService, "127.0.0.1:8081");
            // 绑定服务
            server.bind(helloService);
            // 发布并注册服务
            server.publishServer();
            System.out.println("服务发布成功");
        }
    }

public class ClusterServerService_2 {
        public static void main(String[] args) {
            ISelfHelloService helloService = new SelfHello2Service();
            IRegisterService registerService = new RegisterService();
            // 初始化注册中心和服务端口信息
            RPCServer server = new RPCServer(registerService, "127.0.0.1:8082");
            // 绑定服务
            server.bind(helloService);
            // 发布并注册服务
            server.publishServer();
            System.out.println("服务发布成功");
        }
    }

\* 服务启动执行结果

![70][]

![70 1][]

\* zookeeper查询节点

![70 2][]

### 客户端发起请求 ###

1，jar包依赖

<dependencies>
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.11</version>
          <scope>test</scope>
        </dependency>
        <dependency>
          <groupId>com.gupao.server</groupId>
          <artifactId>RMIServerProject</artifactId>
          <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
          <groupId>org.apache.zookeeper</groupId>
          <artifactId>zookeeper</artifactId>
          <version>3.4.8</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-client</artifactId>
          <version>4.0.0</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.curator</groupId>
          <artifactId>curator-recipes</artifactId>
          <version>4.0.0</version>
        </dependency>
      </dependencies>

2，通过Zookeeper获取注册服务

\* 通过构造方法初始化Curator客户端

public class DiscoveryService implements IDiscoveryService {
    
        private CuratorFramework curatorFramework;
    
        List<String> serviceNodes = new ArrayList<String>();
    
        private String address;
    
        public DiscoveryService(String address){
            curatorFramework = CuratorFrameworkFactory.builder()
                    .connectString(address).sessionTimeoutMs(4000)
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
            curatorFramework.start();
        }

\* 通过serviceName获取服务节点

public String discovery(String serviceName) {
            try {
                // 根据路径获取所有节点
                String serviceUrl = ZKConfig.ZK_REGISTER_PATH + "/" + serviceName;
                serviceNodes = curatorFramework.getChildren().forPath(serviceUrl);
                // 发起服务监听
                registerWatcher(serviceUrl);
                // 实现简易负载均衡
                ILoadBanlanceService loadBanlanceService = new LoadBalanceService();
                return loadBanlanceService.selectHost(serviceNodes);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

\* 服务监听代码

private void registerWatcher(final String path) {
            try {
                PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, path, true);
                PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                        serviceNodes = curatorFramework.getChildren().forPath(path);
                    }
                };
                pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
                pathChildrenCache.start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

\* 通过随机数实现简易负载均衡代码

\* 接口

public interface ILoadBanlanceService {
        String selectHost(List<String> serviceNodes);
    }

\* 抽象类构造模板方法

public abstract class AbstractBalanceService implements ILoadBanlanceService {
        @Override
        public String selectHost(List<String> serviceNodes) {
            if (null == serviceNodes || serviceNodes.size() == 0) {
                return "";
            }
            if (serviceNodes.size() == 1) {
                return serviceNodes.get(0);
            }
            return doSelect(serviceNodes);
        }
    
        public abstract String doSelect(List<String> selectNodes);
    }

\* 实现类实现简易负载均衡

public class LoadBalanceService extends AbstractBalanceService {
    
        @Override
        public String doSelect(List<String> selectNodes) {
            Random random = new Random();
            int indexSelect = random.nextInt(selectNodes.size());
            return selectNodes.get(indexSelect);
        }
    }

3，构建动态代理，发起服务请求

\* 构建动态代理对象

public class RPCClientProxy {
    
        private IDiscoveryService discoveryService;
    
        private String version;
    
        public RPCClientProxy(IDiscoveryService discoveryService, String version) {
            this.discoveryService = discoveryService;
            this.version = version;
        }
    
        public <T> T clientProxy(final Class<T> interafaceCls) {
            return (T) Proxy.newProxyInstance(interafaceCls.getClassLoader(),
                    new Class[]{interafaceCls}, new RemoteInvocationHandler(discoveryService, version));
        }
    }

5，InvocationHandler实现

\* 通过构造方法初始化注册中心对象和版本号

private IDiscoveryService discoveryService;
    
        private String version;
    
        public RemoteInvocationHandler(IDiscoveryService discoveryService, String version) {
            this.discoveryService = discoveryService;
            this.version = version;
        }

\* 动态代理实现方法调用并提交服务请求发送

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            RPCRequest request = new RPCRequest();
            request.setClassName(method.getDeclaringClass().getName());
            request.setMethodName(method.getName());
            request.setParameters(args);
            request.setVersion(version);
            String serviceName = request.getClassName();
            if (null != version && !"".equals(version)) {
                serviceName = serviceName + "-" + version;
            }
            String serviceAddress = discoveryService.discovery(serviceName);
            RPCTransPort transformPort = new RPCTransPort(serviceAddress);
            return transformPort.send(request);
        }

4，发送服务请求到客户端

\* 通过负载均衡获取的服务地址构建套接字

public class RPCTransPort {
    
        private String serviceAddress;
    
        public RPCTransPort(String serviceAddress) {
            this.serviceAddress = serviceAddress;
        }
    
        private Socket newSocket() {
            Socket socket = null;
            try {
                String[] addressArr = serviceAddress.split(":", -1);
                socket = new Socket(addressArr[0], Integer.valueOf(addressArr[1]));
                return socket;
            } catch (Exception e) {
                throw new RuntimeException("CREATED CONNECTION FAILED=======");
            } finally {
            }
        }

\* 通过套接字输出流构建对象输出流

\* 通过对象输出流写出服务请求参数

public Object send(RPCRequest request) {
            Socket socket = null;
            try {
                // 输出服务请求参数到服务端代码======================
                socket = newSocket();
                ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(request);
                outputStream.flush();
                
                // 服务端响应结果代码=========================
                ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                Object result = inputStream.readObject();
                inputStream.close();
                outputStream.close();
                return result;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

5，客户端发送请求main方法

public class ClientController {
        public static void main(String[] args) {
            IDiscoveryService discoveryService = new DiscoveryService(ZKConfig.ZK_HOST);
            for (int i = 0; i < 10; i++) {
                RPCClientProxy rpcClientProxy = new RPCClientProxy(discoveryService, null);
                // 构建代理对象
                ISelfHelloService proxyService =
                        rpcClientProxy.clientProxy(ISelfHelloService.class);
                // 调用服务方法
                String result =  proxyService.sayHello("ZPJ");
                System.out.println(result);
            }
        }
    }

### 服务端响应服务 ###

1，通过服务监听获取客户端请求套接字 （class RPCServer）

public void publishServer() {
            ServerSocket serverSocket = null;
            try {
                // 服务端启动服务代码内容==================
                // 启动服务
                serverSocket = new ServerSocket(Integer.valueOf(serviceAddress.split(":", -1)[1]));
                for (String interfaceName : handlerMap.keySet()) {
                    //
                    registerService.register(interfaceName, serviceAddress);
                    System.out.println("注册服务成功, interfaceName : " + interfaceName + ", serviceAddress : " + serviceAddress);
                }
    
                // 服务端获取客户端请求代码处理======================
                // 获取监听
                while (true) {
                    Socket socket = serverSocket.accept();
                    executorService.execute(new ProcessorHandler(socket, handlerMap));
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (null != serverSocket) {
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }

2，开启服务器线程处理套接字

public class RPCServer {
    
        private final ExecutorService executorService = Executors.newCachedThreadPool();

while (true) {
                    Socket socket = serverSocket.accept();
                    executorService.execute(new ProcessorHandler(socket, handlerMap));
                }

\* 通过构造方法初始化套接字和服务绑定map

public class ProcessorHandler implements Runnable {
    
        private Socket socket;
    
        Map<String, Object> handlerMap;
    
        public ProcessorHandler(Socket socket, Map<String, Object> handlerMap) {
            this.socket = socket;
            this.handlerMap = handlerMap;
        }

\* 获取套接字输入流构建对象输入流

\* 通过输入流读取传输对象

\* 解析对象参数， 执行请求方法并获取结果

public void run() {
            // 处理请求
            ObjectInputStream inputStream = null;
            try {
                // 服务端处理客户端请求代码======================
                inputStream = new ObjectInputStream(socket.getInputStream());
                RPCRequest request = (RPCRequest) inputStream.readObject();
                Object result = invoke(request);
    
                // 服务端返回执行结果到客户端代码========================
                ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(result);
                outputStream.flush();
                outputStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (null != inputStream) {
                    try {
                        inputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

/**
         * 通过执行执行客户端请求方法
         */
        public Object invoke(RPCRequest request) {
            try {
                Object[] parameters = request.getParameters();
                Class<?>[] types = new Class[parameters.length];
                for (int i = 0; i < parameters.length; i++) {
                    types[i] = parameters[i].getClass();
                }
                String version = request.getVersion();
                String serviceName = request.getClassName();
                if (null != version && !"".equals(version)) {
                    serviceName = serviceName + "-" + version;
                }
                Object service = handlerMap.get(serviceName);
                Method method = service.getClass().getMethod(request.getMethodName(), types);
                return method.invoke(service, parameters);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

\* 根据套接字输出流构建对象输出流

\* 通过输出流输出服务执行结果（代码片段）

// 服务端返回执行结果到客户端代码========================
                ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(result);
                outputStream.flush();
                outputStream.close();

### 客户端接收响应 ###

1，根据套接字输入流构建对象输入流（class RPCTransPort）

public Object send(RPCRequest request) {
            Socket socket = null;
            try {
                // 输出服务请求参数到服务端代码======================
                socket = newSocket();
                ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
                outputStream.writeObject(request);
                outputStream.flush();
    
                // 服务端响应结果代码=========================
                ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                Object result = inputStream.readObject();
                inputStream.close();
                outputStream.close();
                return result;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

2，根据对象输入流读取服务端返回执行结果

3，处理返回结果

// 服务端响应结果代码=========================
                ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                Object result = inputStream.readObject();
                inputStream.close();
                outputStream.close();
                return result;

4，请求结果

\* 从下图可以看到，8081和8082端口各有出现；这是因为服务端在注册服务时模拟分布式注册多个服务节点，客户端在处理服务节                 点时进行了简易的基于随机数的负载均衡，main方法中循环调用

\* 请求main方法分十次进行服务获取，每一次都会随机获取8081或者8082端口服务，并进行调用

![70 3][]

## 三，实现多版本处理 ##

1，服务端暴露服务自定义注解上添加version属性

@RpcAnnotation(value = ISelfHelloService.class, version = "1.0")
    public class SelfHelloService implements ISelfHelloService {
    
        @Override
        public String sayHello(String msg) {
            return "[I'M 8081]HELLO : " + msg;
        }
    }

@RpcAnnotation(value = ISelfHelloService.class, version = "2.0")
    public class SelfHello2Service implements ISelfHelloService {
    
        @Override
        public String sayHello(String msg) {
            return "[I'M 8082]HELLO : " + msg;
        }
    }

2，服务绑定根据版本号分别绑定两个服务并注册到注册中心

\* 修改main方法

public class ServerService {
        public static void main(String[] args) {
            ISelfHelloService helloService = new SelfHelloService();
            ISelfHelloService helloService2 = new SelfHello2Service();
            IRegisterService registerService = new RegisterService();
            // 初始化注册中心和服务端口信息
            RPCServer server = new RPCServer(registerService, "127.0.0.1:8080");
            // 绑定服务
            server.bind(helloService, helloService2);
            // 发布服务
            server.publishServer();
            System.out.println("服务发布成功");
        }
    }

\* 注意服务绑定处注册的服务节点为服务名和版本号的组合

3，客户端构建动态代理时，逐层传递要请求的版本号到服务发送端（服务层代码同上）

\* RPCClientProxy第二个参数即为version参数

public class ClientController {
        public static void main(String[] args) {
            IDiscoveryService discoveryService = new DiscoveryService(ZKConfig.ZK_HOST);
            for (int i = 1; i < 3; i++) {
                RPCClientProxy rpcClientProxy = new RPCClientProxy(discoveryService, i + ".0");
                // 构建代理对象
                ISelfHelloService proxyService =
                        rpcClientProxy.clientProxy(ISelfHelloService.class);
                // 调用服务方法
                String result =  proxyService.sayHello("ZPJ");
                System.out.println(result);
            }
        }
    }

\* 再次附上RPCClientProxy代码

4，服务端解析添加版本号处理（服务层代码同上）

\* 重复代码， 为了直观

public Object invoke(RPCRequest request) {
            try {
                Object[] parameters = request.getParameters();
                Class<?>[] types = new Class[parameters.length];
                for (int i = 0; i < parameters.length; i++) {
                    types[i] = parameters[i].getClass();
                }
                String version = request.getVersion();
                String serviceName = request.getClassName();
                if (null != version && !"".equals(version)) {
                    serviceName = serviceName + "-" + version;
                }
                Object service = handlerMap.get(serviceName);
                Method method = service.getClass().getMethod(request.getMethodName(), types);
                return method.invoke(service, parameters);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

5，执行结果

\* 服务端

![70 4][]

\* 客户端

![70 5][]

\* zookeeper节点

![70 6][]

[70]: /images/20220516/1e091c6d47544bf89b0056330bb92b09.png
[70 1]: /images/20220516/bb9ecea5426a45c684bd17df84ad6b41.png
[70 2]: /images/20220516/aa36096c909245ec8f041ebced4397f4.png
[70 3]: /images/20220516/ca55b9ec8b71416aa5874e82ca6c2369.png
[70 4]: /images/20220516/8b193eccf4954eb7bfe491ef59c37ac9.png
[70 5]: /images/20220516/d1c99257c2564cb2abcfb7cb9710a40c.png
[70 6]: /images/20220516/6c52ae94687c41bb8c8d808426064fe0.png