rpc简易实现-zookeeper

　　一、RPC（Remote Procedure Call）—远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

　　RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

　　二、调用过程

　　1.调用客户端句柄；执行传送参数

　　2.调用本地系统内核发送网络消息

　　3.消息传送到远程主机

　　4.服务器句柄得到消息并取得参数

　　5.执行远程过程

　　6.执行的过程将结果返回服务器句柄

　　7.服务器句柄返回结果，调用远程系统内核

　　8.消息传回本地主机

　　9.客户句柄由内核接收消息

　　10.客户接收句柄返回的数据

　　三、调用过程，大概就是这样子，这样子来张图简单一点

　　1、这张图是dubbo官网的模型图

　　2、过程：服务——>注册地址到zk

　　　　　　客户端——>通过zk获取对应服务地址

　　　　　　代理——>通过接口代理，实现服务调用

　　　　　　调用方式——>这个不一定了，一般tcp方式比较直接

　　四、简易版rpc实现，注册中心用的zk，可以使用其他注册中心

　　1）目录结构

　　2）依赖的jar

    　　<dependency>

            <groupId>org.apache.zookeeper</groupId>

            <artifactId>Zookeeper</artifactId>

            <version>3.4.0</version>

        </dependency>

　　3）zookeeper的连接工具

import org.apache.zookeeper.*;

import java.io.IOException;

import java.util.concurrent.CountDownLatch;

public class ZookeeperUtils {

    private static ZooKeeper zooKeeper = null;

    public static ZooKeeper connect() throws IOException, InterruptedException {

        CountDownLatch latch = new CountDownLatch(1);

        //连接zk

        zooKeeper = new ZooKeeper("localhost:2182", 60000, watchedEvent -> {

            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {

                latch.countDown();

            }

        });

        //无连接阻塞

        latch.await();

        return zooKeeper;

    }

}

　　4）数据传输的实体

import java.io.Serializable;

/**

 * 请求需要的数据

 */

public class RpcRequest implements Serializable{

    //接口名称，用于反射

    private String interfaceName;

    //调用方法

    private String method;

    //参数类型

    private Class<?>[] parameterTypes;

    //参数

    private Object[] params;

    public String getInterfaceName() {

        return interfaceName;

    }

    public void setInterfaceName(String interfaceName) {

        this.interfaceName = interfaceName;

    }

    public String getMethod() {

        return method;

    }

    public void setMethod(String method) {

        this.method = method;

    }

    public Class<?>[] getParameterTypes() {

        return parameterTypes;

    }

    public void setParameterTypes(Class<?>[] parameterTypes) {

        this.parameterTypes = parameterTypes;

    }

    public Object[] getParams() {

        return params;

    }

    public void setParams(Object[] params) {

        this.params = params;

    }

}

import java.io.Serializable;

/**

 * 相应对象

 */

public class RpcResponse implements Serializable {

    //返回状态，当然这里可以，加入对应的异常等（这里简化）

    private String status;

    //返回的数据

    private Object data;

    public String getStatus() {

        return status;

    }

    public void setStatus(String status) {

        this.status = status;

    }

    public Object getData() {

        return data;

    }

    public void setData(Object data) {

        this.data = data;

    }

}

　　5）测试的服务以及实现

import com.pinnet.zookeeper.service.ITestService;

public class TestServiceImpl implements ITestService{

    public String test(String name) {

        System.out.println(name);

        return "return" + name;

    }

}

　　6）提供方

import com.pinnet.zookeeper.bean.RpcBeanFactory;

import com.pinnet.zookeeper.data.RpcRequest;

import com.pinnet.zookeeper.data.RpcResponse;

import com.pinnet.zookeeper.service.ITestService;

import com.pinnet.zookeeper.service.impl.TestServiceImpl;

import com.pinnet.zookeeper.zookeeper.ZookeeperUtils;

import org.apache.zookeeper.*;

import java.io.IOException;

import java.io.ObjectInputStream;

import java.io.ObjectOutputStream;

import java.lang.reflect.InvocationTargetException;

import java.lang.reflect.Method;

import java.net.ServerSocket;

import java.net.Socket;

/**

 * rpc服务端

 */

public class Server {

    //服务器设定的目录

    private String registryPath = "/registry";

    //接口，这里方便测试用

    private String serviceName = ITestService.class.getName();

    //地址目录

    private static String addressName = "address";

    //本地地址

    private static String ip = "localhost";

    //监听接口

    public static Integer port = 8000;

    //链接zk

    public ZooKeeper connect() throws Exception {

        ZooKeeper zooKeeper = ZookeeperUtils.connect();

        return zooKeeper;

    }

    //创建节点，也就是访问的，目录

    public void createNode(ZooKeeper zooKeeper) throws Exception {

        if (zooKeeper.exists(registryPath, false) == null) {

            //创建永久目录，接口服务，可以创建永久目录

            zooKeeper.create(registryPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        }

        String servicePath = registryPath + "/" +serviceName;

        if (zooKeeper.exists(servicePath, false) == null) {

            //接口目录

            zooKeeper.create(servicePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        }

        String addressPath = servicePath + "/" +addressName;

        //地址目录，这里ip就是本地的地址，用于tcp链接使用

        //这里创建的是临时目录，当zk服务断连过后，自动删除临时节点

        zooKeeper.create(addressPath, (ip + ":"+ port).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

    }

    //监听过程

    private void accept() throws Exception {

        //当然这里也可以使用netty来进行监听和其他过程

        //这里简化

        ServerSocket serverSocket = new ServerSocket(port);

        while (true) {

            System.out.println("监听中。。。。。。。");

            Socket socket = serverSocket.accept();

            resultData(socket);

        }

    }

    //执行并返回数据

    private void resultData(Socket socket) throws IOException, ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {

        ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());

        //读取请求的参数

        RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();

        //这里从容器中获取bean，当然这里的bean可以自己缓存，独立spring容器之外

        Object bean = RpcBeanFactory.getBean(rpcRequest.getInterfaceName());

        //方法调用

        Method method = bean.getClass().getMethod(rpcRequest.getMethod(), rpcRequest.getParameterTypes());

        Object data = method.invoke(bean, rpcRequest.getParams());

        //返回数据

        ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());

        RpcResponse rpcResponse = new RpcResponse();

        rpcResponse.setStatus("success");

        rpcResponse.setData(data);

        objectOutputStream.writeObject(rpcResponse);

    }

    public static void main(String[] args) throws Exception {

        //模拟spring容器的加载过程

        RpcBeanFactory.putBean(ITestService.class.getName(), new TestServiceImpl());

        Server server = new Server();

        ZooKeeper zooKeeper = server.connect();

        //创建节点，用于地址访问

        server.createNode(zooKeeper);

        //监听，当然多线程更加理想，这里只显示效果

        server.accept();

    }

}

　　7）调用方

import com.pinnet.zookeeper.bean.RpcBeanFactory;

import com.pinnet.zookeeper.data.RpcRequest;

import com.pinnet.zookeeper.data.RpcResponse;

import com.pinnet.zookeeper.service.ITestService;

import com.pinnet.zookeeper.zookeeper.ZookeeperUtils;

import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;

import java.io.ObjectInputStream;

import java.io.ObjectOutputStream;

import java.lang.reflect.InvocationHandler;

import java.lang.reflect.Method;

import java.lang.reflect.Proxy;

import java.net.InetSocketAddress;

import java.net.Socket;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.concurrent.ConcurrentHashMap;

/**

 * rpc客户端

 */

public class Client {

    //缓存地址

    private Map<String, List<String>> adressMap = new ConcurrentHashMap<>();

    //链接zk

    public ZooKeeper connect() throws Exception {

        return ZookeeperUtils.connect();

    }

    /**

     * 这里主要是创建代理，利用代理接口实现来达到调用的目的

     * @param interfaceName

     * @return

     * @throws ClassNotFoundException

     */

    public Object createProxy(final String interfaceName) throws ClassNotFoundException {

        //使用线程实例化，主要考虑处理性

        final Class clazz = Thread.currentThread().getContextClassLoader().loadClass(interfaceName);

        //创建代理

        return Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new InvocationHandler() {

            //用重连计数

            private int num = 5;

            @Override

            public Object invoke(Object proxy, Method method, Object[] params) throws Throwable {

                //发送请求需要的请求参数

                RpcRequest rpcRequest = new RpcRequest();

                //接口名称

                rpcRequest.setInterfaceName(interfaceName);

                //调用方法名

                rpcRequest.setMethod(method.getName());

                //对应参数类型

                rpcRequest.setParameterTypes(method.getParameterTypes());

                //对应参数

                rpcRequest.setParams(params);

                //返回响应结果

                return sendData(rpcRequest);

            }

            //tcp方式调用

            private Object sendData(RpcRequest rpcRequest) throws Exception {

                //当访问地址存在时

                if (adressMap.containsKey(interfaceName)) {

                    List<String> adresses = adressMap.get(interfaceName);

                    if (adresses != null && !adresses.isEmpty()) {

                        //如果存在多个地址，使用可以调通的一个

                        for (String adress:adresses) {

                            //这个是注册zk的时候设定的数据

                            String[] strs = adress.split(":");

                            //这里简易版的实现，所以直接使用的socket.

                            //实际上可以采用netty框架编写，保存channel访问就可以了，也可以对数据进行加解密

                            Socket socket = new Socket();

                            try {

                                //连接可以访问zk，如果连接失败直接抛出异常，循环下一个地址

                                socket.connect(new InetSocketAddress(strs[0], Integer.valueOf(strs[1])));

                                ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());

                                objectOutputStream.writeObject(rpcRequest);

                                ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());

                                //这里是响应数据，也就是执行过后的数据。远程执行结果

                                RpcResponse rpcResponse = (RpcResponse) objectInputStream.readObject();

                                return rpcResponse.getData();

                            } catch (IOException e) {

                                e.printStackTrace();

                            }

                        }

                        //设置重连机制，跳出循环

                        if (num == 0) {

                            throw new RuntimeException("server connect fail");

                        }

                        num--;

                        //如果多个地址还是不能访问，则从zk上面更新地址

                        getAddress(interfaceName);

                        //如果多个地址还是不能访问，则重新访问

                        sendData(rpcRequest);

                    }

                    throw new RuntimeException("not found service");

                }else {

                    //如果没有地址存储时的访问，主要是请求链接问题

                    if (num == 0) {

                        throw new RuntimeException("not found server");

                    }

                    num--;

                    getAddress(interfaceName);

                    return sendData(rpcRequest);

                }

            }

        });

    }

    //从zk获取最新的地址

    private void getAddress(String interfaceName) throws Exception {

        //链接zk

        ZooKeeper zooKeeper = connect();

        //设定的地址目录

        String interfacePath = "/registry/" + interfaceName;

        //获取地址目录

        List<String> addresses = zooKeeper.getChildren(interfacePath, false);

        if (addresses != null && !addresses.isEmpty()) {

            List<String> datas = new ArrayList<>();

            for (String address:addresses) {

                //获取数据，也就是配置对应的访问地址

                byte[] bytes = zooKeeper.getData(interfacePath + "/" + address, false, null);

                if (bytes.length > 0) {

                    //放入数组

                    datas.add(new String(bytes));

                }

            }

            //加入缓存

            adressMap.put(interfaceName, datas);

        }

    }

    public static void main(String[] args) throws ClassNotFoundException, InterruptedException {

        Client client = new Client();

        //这一步是模拟spring容器放入bean的过程，实际用spring容器可自定义标签实现

        RpcBeanFactory.putBean("testService", client.createProxy(ITestService.class.getName()));

        //获取bean的过程，实际上实现是代理接口的实现

        ITestService testService = (ITestService) RpcBeanFactory.getBean("testService");

        //调用方法，也就是调用代理的过程

        String test = testService.test("test");

        System.out.println(test);

    }

}

　　五、代码部分就这么多，看一下测试结果

　　服务端：

　　客户端：

　　zookeeper的展示部分

　　六、源码分享

　　rpc-zk代码：https://github.com/lilin409546297/rpc-zk

　　zkui代码：https://github.com/lilin409546297/zkui（这个是别人做的，我放在github的）

　　zookeeper下载地址：http://apache.org/dist/zookeeper/

巴特西

rpc简易实现-zookeeper

最新文章

热门文章