当前位置：首页 > news >正文

简单的实现RPC框架

news 来源：原创 2025/5/16 19:38:53

首先我们要知道RPC框架是啥

1 RPC
1.1 RPC概念
RPC（Remote Procedure Call Protocol）远程过程调用协议。
RPC是一种通过网络从远程计算机程序上请求服务。
RPC主要作用就是不同的服务间方法调用就像本地调用一样便捷。
1.2 常用RPC技术或框架
应用级的服务框架：阿里的 Dubbo/Dubbox、Google gRPC、Spring Boot/Spring Cloud。
远程通信协议：RMI、Socket、SOAP(HTTP XML)、REST(HTTP JSON)。
通信框架：MINA 和 Netty

1.3 初始化工程：

Consumer（消费者）、Producer（生产者）、Producer-common（生产者接口）、SLQF-rpc（RPC模块）

在生产者接口模块里面创建一个接口HelloService

package com.producercommon.Service;

public interface HelloService {
    String sayhello(String name);
}

然后在生产者和消费者模块中都导入生产者接口和RPC模块。

<dependency>
     <groupId>com</groupId>
     <artifactId>SLQF-rpc</artifactId>
     <version>0.0.1-SNAPSHOT</version>
</dependency>

<dependency>
     <groupId>com</groupId>
     <artifactId>Producer-common</artifactId>
     <version>0.0.1-SNAPSHOT</version>
</dependency>

在生产者模块实现HelloService接口

package com.producer.Impl;

import com.producercommon.Service.HelloService;

public class HelloServiceImpl implements HelloService {
    @Override
    public String sayhello(String name) {
        return "hello:" + name;
    }
}

编写RPC框架
在消费者模块中，不能通过创建HelloServiceImpl对象来调用Producer中的sayhello方法，那就只能通过网络请求的方式来调用，接收网络请求的方式有很多
如Jetty，Tomcat，Netty，或者直接用Socket，这里选择使用的是Tomcat
在RPC模块中导入Tomcat依赖以及IO工具包

<dependency>
      <groupId>org.apache.tomcat.embed</groupId>
      <artifactId>tomcat-embed-core</artifactId>
      <version>9.0.79</version>
</dependency>

<dependency>
      <groupId>commons-io</groupId>
      <artifactId>commons-io</artifactId>
      <version>1.3.1</version>
</dependency>

需要在Producer服务启动时，启动Tomcat，在RPC模块中编写一个Tomcat启动方法，方法中的参数为IP地址和绑定的端口号

package com.youyuan;

import org.apache.catalina.*;
import org.apache.catalina.connector.Connector;
import org.apache.catalina.core.StandardContext;
import org.apache.catalina.core.StandardEngine;
import org.apache.catalina.core.StandardHost;
import org.apache.catalina.startup.Tomcat;

public class HttpServer {
    public void start(String hostName, int port) {
        Tomcat tomcat = new Tomcat();

        Server server = tomcat.getServer();
        Service service = server.findService("Tomcat");

        Connector connector = new Connector();
        connector.setPort(port);

        Engine engine = new StandardEngine();
        engine.setDefaultHost(hostName);

        Host host = new StandardHost();
        host.setName(hostName);

        String contextPath = "";
        Context context = new StandardContext();
        context.setPath(contextPath);
        context.addLifecycleListener(new Tomcat.FixContextListener());

        host.addChild(context);
        engine.addChild(host);

        service.setContainer(engine);
        service.addConnector(connector);

        // 将所有请求都交给dispatcher处理
        tomcat.addServlet(contextPath, "dispatcher", new DispatcherServlet());
        context.addServletMappingDecoded("/*", "dispatcher");

        try {
            tomcat.start();
            tomcat.getServer().await();
        } catch (LifecycleException e) {
            e.printStackTrace();
        }
    }
}

其中使用的dispatcher为自定义的DispatcherServlet，用来处理收到的网络消息
同样在RPC模块中编写DispatcherServlet类

package com.youyuan;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;

public class DispatcherServlet extends HttpServlet {
    @Override
    protected void service(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        // 抽象为Hander类
        new HttpServerHandler().handler(req, resp);
    }
}

将处理方式抽象为一个类，提升框架可扩展性，如果还要处理其他类型的网络消息，可以增加另外的Handler类，随后编写客户端发送的Invocation消息类

package com.youyuan;

import java.io.Serializable;

public class Invocation implements Serializable {
    // 接口名称
    private String interfaceName;
    // 需要调用的方法名
    private String methodName;
    // 方法中参数类型
    private Class[] types;
    // 方法的参数
    private Object[] args;

    public String getInterfaceName() {
        return interfaceName;
    }

    public void setInterfaceName(String interfaceName) {
        this.interfaceName = interfaceName;
    }

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    public Class[] getTypes() {
        return types;
    }

    public void setTypes(Class[] types) {
        this.types = types;
    }

    public Object[] getArgs() {
        return args;
    }

    public void setArgs(Object[] args) {
        this.args = args;
    }

    public Invocation(String interfaceName, String methodName, Class[] types, Object[] args) {
        this.interfaceName = interfaceName;
        this.methodName = methodName;
        this.types = types;
        this.args = args;
    }
}

之后是发送网络消息的客户端方法，参数为IP地址，Tomcat服务器端口号，消息Invocation

package com.youyuan;


import org.apache.commons.io.IOUtils;

import java.io.InputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;

public class HttpClient {
    public String send(String host, int port, Invocation invocation) {
        String result = null;
        try {
            URL url = new URL("http", host, port, "/");
            HttpURLConnection httpConnection = (HttpURLConnection) url.openConnection();

            httpConnection.setRequestMethod("POST");
            httpConnection.setDoOutput(true);

            OutputStream out = httpConnection.getOutputStream();
            ObjectOutputStream outStream = new ObjectOutputStream(out);
            outStream.writeObject(invocation);
            outStream.flush();
            outStream.close();

            InputStream inputStream = httpConnection.getInputStream();
            result =  IOUtils.toString(inputStream);
            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }
}

当收到客户端发送网络消息后，Tomcat默认调用我们配置的DispatcherServlet，编写其中处理消息的Handler类

package com.youyuan;


import org.apache.commons.io.IOUtils;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.ObjectInputStream;
import java.lang.reflect.Method;

public class HttpServerHandler {
    public void handler(HttpServletRequest request, HttpServletResponse response){
        try {
            // 获取客户端传过来的Invocation消息
            Invocation invocation = (Invocation) new ObjectInputStream(request.getInputStream()).readObject();
            String interfaceName = invocation.getInterfaceName();
            String methodName = invocation.getMethodName();
            // 根据接口名称，获取想要调用的实现类
            Class clazz = LocalRegister.get(interfaceName);
            Method method = clazz.getMethod(methodName, invocation.getTypes());
            // 利用反射执行消费者想要调用的方法
            String result = (String)method.invoke(clazz.newInstance(), invocation.getArgs());
            // 返回方法的执行结果
            IOUtils.write(result, response.getOutputStream());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这就是处理消息的类，也是调用服务关键所在，以下代码段中的LocalRegister类，是一个本地注册类，里面保存了接口与对应实现类的映射，保证调用的是想要的实现类的方法，以下是LocalRegister的代码

package com.youyuan;

import java.util.*;

public class LocalRegister {
    private static Map<String, Class> map = new HashMap<>();

    public static void register(String interfaceName, Class implClass) {
        map.put(interfaceName, implClass);
    }

    public static Class get(String ingerfaceName) {
        return map.get(ingerfaceName);
    }
}

RPC框架通常返回的是一个代理对象，然后来执行代理对象的方法来实现服务的调用，接下来我们来编写一下代理类

package com.youyuan;


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

public class ProxyFactory {
    public static <T>T getProxy(Class interfaceClass) {
        Object proxy = Proxy.newProxyInstance(ProxyFactory.class.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 使用反射来创建消息对象
                Invocation invocation = new Invocation(interfaceClass.getName(), method.getName(),
                        method.getParameterTypes(), args);
                HttpClient client = new HttpClient();
                String result = client.send("localhost", 8080, invocation);
                return result;
            }
        });
        return (T)proxy;
    }
}

现在RPC框架的主要架构已经编写完成，可以进行测试一下，在Producer服务模块中启动Tomcat并且注册服务

package com.producer.Impl;


import com.producercommon.Service.HelloService;
import com.youyuan.HttpServer;
import com.youyuan.LocalRegister;

public class Producer {
    public static void main(String[] args) {
        // 注册服务
        LocalRegister.register(HelloService.class.getName(), HelloServiceImpl.class);

        HttpServer server = new HttpServer();
        server.start("localhost",8080);
    }
}

Consumer模块创建HelloService的代理对象来调用Producer中的服务了

package com.consumer;


import com.producercommon.Service.HelloService;
import com.youyuan.ProxyFactory;

public class Consumer {
    public static void main(String[] args) {
        HelloService helloService = ProxyFactory.getProxy(HelloService.class);
        String result = helloService.sayhello("小白鼬猿");
        System.out.println(result);
    }
}

执行代码，发现调用成功啦~：