Java-Handschrift eines RPC-Frameworks

Das RPC-Framework wird als Remote-Calling-Framework bezeichnet. Das Kernprinzip seiner Implementierung besteht darin, dass die Verbraucherseite einen dynamischen Proxy verwendet, um eine Schnittstelle zu vertreten (Dynamischer Proxy, der natürlich auf JDK basiert). Wenn Sie CGLib verwenden, können Sie Methoden ohne Schnittstellenklassen direkt verwenden ). Anschließend wird die Reflexion ausgeführt. Die Ergebnisse werden über die Netzwerkprogrammierung an den Verbraucher zurückgesendet.

Lassen Sie uns nun diese Konzepte der Reihe nach umsetzen. Hier führen wir die einfachste Implementierung durch. Sie können Netty im Reactor-Modus verwenden, um eine bessere Leistung zu erzielen. Die bei der Netzwerkübertragung verwendete Serialisierung und Deserialisierung sind ebenfalls in Java enthalten. Natürlich sind solche Übertragungsbytes relativ groß und können mit Googles ProtoBuffer oder Kryo verarbeitet werden. Dies dient nur der Vereinfachung, um das Prinzip zu erklären.

pom

<?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>com.guanjian</groupId>
    <artifactId>rpc-framework</artifactId>
    <version>1.0-SNAPSHOT</version>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.7.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Das erste ist natürlich die Schnittstelle, die wir remote aufrufen möchten, und die Schnittstellenmethode.

public interface HelloService {
    String sayHello(String content);}

Schnittstellenimplementierungsklasse

public class HelloServiceImpl implements HelloService {    public String sayHello(String content) {        return "hello," + content;    }
}

Dynamischer Proxy auf der Verbraucherseite, Wenn Sie den Anbieter und den Verbraucher in zwei Projekten schreiben, benötigt die Anbieterseite die oben genannten Schnittstellen- und Implementierungsklassen, während Die Verbraucherseite benötigt nur die obige Schnittstelle.

public class ConsumerProxy {
    /**
     * 消费者端的动态代理
     * @param interfaceClass 代理的接口类
     * @param host 远程主机IP
     * @param port 远程主机端口
     * @param <T>
     * @return
     */
    @SuppressWarnings("unchecked")
    public static <T> T consume(final Class<T> interfaceClass,final String host,final int port) {
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                new Class[]{interfaceClass}, (proxy,method,args) -> {
                    //创建一个客户端套接字
                    Socket socket = new Socket(host, port);
                    try {
                        //创建一个对外传输的对象流，绑定套接字
                        ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                        try {
                            //将动态代理的方法名写入对外传输的对象流中
                            output.writeUTF(method.getName());
                            //将动态代理的方法的参数写入对外传输的对象流中
                            output.writeObject(args);
                            //创建一个对内传输的对象流，绑定套接字
                            //这里是为了获取提供者端传回的结果
                            ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                            try {
                                //从对内传输的对象流中获取结果
                                Object result = input.readObject();
                                if (result instanceof Throwable) {
                                    throw (Throwable) result;
                                }
                                return result;
                            } finally {
                                input.close();
                            }
                        } finally {
                            output.close();
                        }
                    } finally {
                        socket.close();
                    }
                }
        );
    }
}

Informationen zum dynamischen JDK-Proxy finden Sie unter AOP-Prinzipien und Selbstimplementierung. Für BIO lesen Sie bitte den Vergleich zwischen traditionellem IO und NIO Netzwerkübertragungs- und Fernanrufdienste

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

    /**
     * RPC监听和远程方法调用
     * @param service RPC远程方法调用的接口实例
     * @param port 监听的端口
     * @throws Exception
     */
    public static void provider(final Object service,int port) throws Exception {
        //创建服务端的套接字，绑定端口port
        ServerSocket serverSocket = new ServerSocket(port);
        while (true) {
            //开始接收客户端的消息，并以此创建套接字
            final Socket socket = serverSocket.accept();
            //多线程执行，这里的问题是连接数过大，线程池的线程数会耗尽
            executorService.execute(() -> {
                try {
                    //创建呢一个对内传输的对象流，并绑定套接字
                    ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                    try {
                        try {
                            //从对象流中读取接口方法的方法名
                            String methodName = input.readUTF();
                            //从对象流中读取接口方法的所有参数
                            Object[] args = (Object[]) input.readObject();
                            Class[] argsTypes = new Class[args.length];
                            for (int i = 0;i < args.length;i++) {
                                argsTypes[i] = args[i].getClass();

                            }
                            //创建一个对外传输的对象流，并绑定套接字
                            //这里是为了将反射执行结果传递回消费者端
                            ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                            try {
                                Class<?>[] interfaces = service.getClass().getInterfaces();
                                Method method = null;
                                for (int i = 0;i < interfaces.length;i++) {
                                    method = interfaces[i].getDeclaredMethod(methodName,argsTypes);
                                    if (method != null) {
                                        break;
                                    }
                                }
                                Object result = method.invoke(service, args);
                                //将反射执行结果写入对外传输的对象流中
                                output.writeObject(result);
                            } catch (Throwable t) {
                                output.writeObject(t);
                            } finally {
                                output.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            input.close();
                        }
                    } finally {
                        socket.close();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

Netzwerküberwachung und Fernanrufe auf der Anbieterseite starten

public class RPCProviderMain {
    public static void main(String[] args) throws Exception {
        HelloService service = new HelloServiceImpl();
        ProviderReflect.provider(service,8083);
    }
}

Dynamische Proxy-Anrufe beim Verbraucher starten

public class RPCConsumerMain {
    public static void main(String[] args) throws InterruptedException {
        HelloService service = ConsumerProxy.consume(HelloService.class,"127.0.0.1",8083);
        for (int i = 0;i < 1000;i++) {
            String hello = service.sayHello("你好_" + i);
            System.out.println(hello);
            Thread.sleep(1000);
        }
    }
}

Ergebnisse ausführen

Hallo, hallo _0

Hallo, hallo_2
Hallo, hallo_3
Hallo, hallo_4
Hallo, hallo_5

.....

Wenn Sie es mit Netty + ProtoBuffer zu einem leistungsstarken RPC-Framework erweitern möchten, können Sie sich auf die entsprechenden Schreibmethoden zur Integration von Protobuffer durch Netty beziehen.

Empfohlenes Tutorial: „

“

Das obige ist der detaillierte Inhalt vonJava-Handschrift eines RPC-Frameworks. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!