重点在理rpc的核心原理

image.png

1.服务端

image.png

1.1 核心服务器与反射调用

package com.gzz.server;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.gzz.common.RpcRequest;
import com.gzz.common.RpcResponse;
import com.gzz.user.UserService;
import com.gzz.user.UserServiceImpl;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class AppService {
    private static final int PORT = 8080;
    private static ExecutorService executorService = new ThreadPoolExecutor(5, 10, 3, TimeUnit.MINUTES, new ArrayBlockingQueue<>(10));
    private static ServerSocket serverSocket = null;
    public static void main(String[] args) {
        UserService userService = new UserServiceImpl();
        log.info("服务器已启动, 端口为:{}", PORT);
        try {
            serverSocket = new ServerSocket(PORT);
            for (;;) {
                // socket为客户端的一个引用，到该方法会连接阻塞
                Socket socket = serverSocket.accept();
                // 使用线程池来处理请求，避免阻塞
                executorService.submit(() -> invoke(socket, userService));
            }
        } catch (IOException e) {
            log.error("", e);
        }
    }
    public static void invoke(Socket socket, Object instance) {
        try {
            // 客户端的输入，其实就是RpcRequest对象
            RpcRequest request = (RpcRequest) new ObjectInputStream(socket.getInputStream()).readObject();
            // 参数的类信息，用于指定方法获取
            Object[] params = request.getParams();
            // 参数类型数组
            Class<?>[] paramsType = Arrays.asList(params).stream().map(p -> p.getClass()).collect(Collectors.toList()).toArray(new Class<?>[params.length]);
            Object result = Class.forName(request.getClassName()).getMethod(request.getMethodName(), paramsType).invoke(instance, params);
            // 到输出流，把返回结果写出到客户端
            new ObjectOutputStream(socket.getOutputStream()).writeObject(RpcResponse.success(result));
        } catch (Exception e) {
            log.error("", e);
        }
    }
}

1.2 com.gzz.user.User

package com.gzz.user;
import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class User implements Serializable {
    private static final long serialVersionUID = 1L;
    private Integer id;// uid
    private String name;// 用户名
    private Integer age;// 年龄
}

1.3 com.gzz.user.UserService接口

package com.gzz.user;
import java.util.List;
public interface UserService {
    public void save(User user);
    public List<User> list(Integer id);
}

1.4 com.gzz.user.UserServiceImpl实现类

package com.gzz.user;
import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class UserServiceImpl implements UserService {
    @Override
    public void save(User user) {
        log.info("user={}", user);
    }
    @Override
    public List<User> list(Integer id) {
        List<User> users = new ArrayList<>();
        users.add(User.builder().name("zhangsan").age(32).id(1).build());
        users.add(User.builder().name("lisi").age(22).id(3).build());
        return users;
    }
}

1.5 com.gzz.common.RpcRequest 请求包装类

package com.gzz.common;
import java.io.Serializable;
import lombok.Data;
@Data
public class RpcRequest implements Serializable {
    private static final long serialVersionUID = 1L;
    // 类的全路径
    private String className;
    // 方法的名称
    private String methodName;
    // 方法的参数
    private Object[] params;
}

1.6 com.gzz.common.RpcResponse响应包装类

package com.gzz.common;
import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class RpcResponse implements Serializable {
    private static final long serialVersionUID = 1L;
    private Integer code;
    private String message;
    private Object result;
    public static RpcResponse success(Object result) {
        return new RpcResponse(200, "结果正确", result);
    }
    public static RpcResponse failed(String message) {
        return new RpcResponse(500, message, null);
    }
}

2.客户端

image.png

2.1核心的动态代理与反射

package com.gzz.client;
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.Socket;
import com.gzz.common.RpcRequest;
import com.gzz.common.RpcResponse;
import com.gzz.user.User;
import com.gzz.user.UserService;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class AppClient {
    public static void main(String[] args) {
        // 创建一个代理对象，对接口进行代理，获取代理对象,当调用方法的时候，其实是执行代理对象的invoke方法
        UserService userService = (UserService) Proxy.newProxyInstance(AppClient.class.getClassLoader(), new Class[] { UserService.class }, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 只需要封装好RpcRequest对象，传给服务端即可
                RpcRequest request = RpcRequest.builder().className(method.getDeclaringClass().getName()).methodName(method.getName()).params(args).build();
                Socket socket = new Socket("localhost", 8080);
                // 获取到输出流，把rpcRequest对象传到服务端，让服务端进行解析
                new ObjectOutputStream(socket.getOutputStream()).writeObject(request);
                // 获取到输入流，从服务器端获取到执行的结果，并对其进行解析
                RpcResponse response = (RpcResponse) new ObjectInputStream(socket.getInputStream()).readObject();
                socket.close();
                return response.getResult();
            }
        });
        // 获取id为10的用户
        log.info("users={}", userService.list(1));
        // 执行保存操作
        userService.save(User.builder().name("").age(44).id(10).build());
    }
}

2.2 另外四个类与服务端相同

2.3 pom.xml

<?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.gzz</groupId>
    <artifactId>11-rpc-server</artifactId>
    <version>1.0</version>
    <packaging>jar</packaging>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.6</version>
    </parent>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>