手撕RPC实现基于TCP的仿Dubbo简单实现

瑞新

发布于 2020-07-07 11:36:43

7330

发布于 2020-07-07 11:36:43

文章被收录于专栏：用户3288143的专栏

文章目录

手撕RPC实现基于TCP的仿Dubbo实现
- 方法调用效果实现
- 分模块
- 写接口
- 通过代理写好了客户端
- 写服务端，并联调rpc
- 代码实现
  - myRpc
  - rpc-client
  - rpc-interface
  - rpc-server
  - 源码

手撕RPC实现基于TCP的仿Dubbo实现

还记得第一次听到这词是在别人的面试视频里，简单了解了一下只知道是远程调用。万万没想到我的第一次面试的第一个问题就是与此相关，希望认真准备每一次面试，及时查漏补缺，谨以此文，代表诚意~奥利给！
思路： my-rpc通过client调用interface给server

方法调用效果实现

分模块

写接口

序列化、并统一编码

实例和接口

通过Socket实现Rpc，注意协调模块间依赖首先实现服务端服务端：方法实现需要依赖接口的对象实例客户端：UserInfoService.class需要依赖接口提供 myRpc：需要给客户端、服务端提供服务

通过代理写好了客户端

运行测试连接拒绝因为远程服务还没有开 java.net.ConnectException: Connection refused (Connection refused)

写服务端，并联调rpc

时刻提醒自己保持tcp下配合反射，代理

代码实现

myRpc

ServerHandler

package com.bennyrhys.handler;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * 对服务端的请求处理
 * @Author bennyrhys
 * @Date 2020-03-22 09:58
 */
public class ServerHandler {
    /**
     * 对请求的处理
     * @param target
     * @param port
     */
    public void handel(Object target, int port) {
        try {
            // 接收后端连接
            ServerSocket serverSocket = new ServerSocket(port);

            while (true) {
                // 阻塞方法
                Socket socket = serverSocket.accept();

                // 运行一个线程
                // 接收远程请求，对请求进行处理
                new Thread(new ServiceProcess(socket, target)).start();
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

ServiceProcess

package com.bennyrhys.handler;

import com.bennyrhys.request.RpcRequest;

import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.net.Socket;

/**
 * 线程服务的处理
 * @Author bennyrhys
 * @Date 2020-03-22 10:04
 */
public class ServiceProcess implements Runnable {

    private Socket socket;
    // 目标对象，也就是UserServiceImpl
    private Object target;

    public ServiceProcess(Socket socket, Object target) {
        this.socket = socket;
        this.target = target;
    }



    public void run() {

        try {
            // 拿到客户端发送来的数据
            InputStream is = socket.getInputStream();
            // 包装成对象输入流
            ObjectInputStream ois = new ObjectInputStream(is);
            // 拿到客户端封装的对象
            RpcRequest rpcRequest =  (RpcRequest)ois.readObject();

            // 开始调用service实现-反射
            Object obj = invoke(rpcRequest);

            OutputStream os = socket.getOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(os);
            oos.writeObject(obj);
            oos.flush();

            os.close();
            oos.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 实现对UserServiceImpl方法getUserInfo()的调用
     * @param rpcRequest
     * @return
     */
    public Object invoke(RpcRequest rpcRequest) {

        String methodName = rpcRequest.getMethodName();
        Object[] parameters = rpcRequest.getParameters();

        Class[] parameterTypes = new Class[parameters.length];
        for (int i = 0; i < parameters.length; i++) {
            parameterTypes[i] = parameters[i].getClass();
        }
        
        try {
            Method  method = target.getClass().getDeclaredMethod(methodName, parameterTypes);
            // 反射调用 目标对象，方法参数
            // methed 代表getUserInfo()方法的反射方法java.long.reflect.Method信息
            // invoke 代表调用getUserInfo()
            // target 你现在要调哪个对象的getUserInfo()方法，要调UserServiceImpl对象，target代表UserServiceImpl的方法
            // parameters 代表getUserInfo()方法的参数，如果方法没有参数，那么就null
            Object obj = method.invoke(target, parameters);
            return obj;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }
}

MyInvocationHandler

package com.bennyrhys.proxy;

import com.bennyrhys.request.RpcInvoke;
import com.bennyrhys.request.RpcRequest;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;

/**
 * 实现InvocationHandler接口
 * 调用getProxyObject方法的时候，会被该invoke拦截
 * @Author bennyrhys
 * @Date 2020-03-22 00:15
 */
public class MyInvocationHandler implements InvocationHandler {

    private String ip;
    private int port;

    public MyInvocationHandler(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }

    /**
     *
     * @param proxy --> 代理类。也就是UserService的代理类
     * @param method --> getUserInfo()
     * @param args --> getUserInfo()方法的参数
     * @return
     * @throws Throwable
     */
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        // 实现对远程方法的调用，先准备数据
        System.out.println("远程调用在此处拦截，进行数据封装");

        String className = method.getDeclaringClass().getName();
        String methodName = method.getName();
        // 封装的对象
        // 避免空指针
        RpcRequest rpcRequest = new RpcRequest(className, methodName, args == null ? new Object[]{} : args);

        // 调用远程方法
        RpcInvoke rpcInvoke = new RpcInvoke(ip, port);
        return rpcInvoke.invoke(rpcRequest);

    }
}

RpcProxy

package com.bennyrhys.proxy;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;

/**
 * rpc代理类
 * @Author bennyrhys
 * @Date 2020-03-22 00:05
 */
public class RpcProxy {

    @SuppressWarnings("unchecked")
    public <T> T getProxyObject(Class<T> classInterFaces, String ip, int port) {
        // 使用JDK去实现动态代理
        return (T)Proxy.newProxyInstance(classInterFaces.getClassLoader(),
                new Class<?>[] {classInterFaces},
                new MyInvocationHandler(ip, port));
    }
}

rpc-client

package com.bennyrhys;

import com.bennyrhys.model.UserInfo;
import com.bennyrhys.proxy.RpcProxy;
import com.bennyrhys.service.UserInfoService;

/**
 * 客户端获取
 * @Author bennyrhys
 * @Date 2020-03-21 23:20
 */
public class client {

    private static final String IP = "127.0.0.1";
    private static final int PORT = 12345;

    public static void main(String[] args) {

        // rpc代理
        RpcProxy proxy = new RpcProxy();

        // 如果拿到UserInfoService的实现类（代理技术）
        UserInfoService userInfoService = proxy.getProxyObject(UserInfoService.class, IP, PORT);
        UserInfo user = userInfoService.getUserInfo();
        System.out.println(user);
    }
}

rpc-interface

package com.bennyrhys.model;

import java.io.Serializable;

/**
 * @Author bennyrhys
 * @Date 2020-03-21 23:04
 */
// Tcp网络传输，要变为序列化
public class UserInfo implements Serializable {
    private Integer id;
    private String name;

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "UserInfo{" +
                "id=" + id +
                ", name='" + name + '\'' +
                '}';
    }
}

UserInfoService

package com.bennyrhys.service;

import com.bennyrhys.model.UserInfo;

/**
 * @Author bennyrhys
 * @Date 2020-03-21 22:47
 */
public interface UserInfoService {
    public UserInfo getUserInfo();
}

rpc-server

UserServiceImpl

package com.bennyrhys.service.impl;

import com.bennyrhys.model.UserInfo;
import com.bennyrhys.service.UserInfoService;

/**
 * 实现用户接口
 * 通过添加依赖
 * @Author bennyrhys
 * @Date 2020-03-21 23:23
 */
public class UserServiceImpl implements UserInfoService {
    public UserInfo getUserInfo() {
        // 写死-假设数据库拿到数据
        UserInfo user = new UserInfo();
        user.setId(10086);
        user.setName("查询");
        return user;
    }
}

Server

package com.bennyrhys;

import com.bennyrhys.handler.ServerHandler;
import com.bennyrhys.service.impl.UserServiceImpl;

import java.net.ServerSocket;

/**
 * 服务端接收
 * @Author bennyrhys
 * @Date 2020-03-22 09:48
 */
public class Server {
    public static void main(String[] args) {
        ServerHandler serverHandler = new ServerHandler();
        // new UserServiceImpl()，可以用注册中心写活， 端口和客户端给定的一样
        serverHandler.handel(new UserServiceImpl(), 12345);
    }
}