从零开始，手写一个RPC，跟随着这篇文档以及数个迭代版本的代码，由简陋到逐渐完备，让所有人都能看懂并且写出一个RPC框架。
本文档与代码都是本人第一次手写RPC的心路历程，会有理解的偏差与代码上的不完善，但更是由于这样，有着与新手对同样问题的疑惑，也许会使新手更容易理解这样做的缘故是啥。
另外期待与你的合作：代码，帮助文档甚至rpc框架功能的完备
学习建议：
- 一定要实际上手敲代码
- 每一版本都有着对应独立的代码与文档，结合来看
- 每一版本前有一个背景知识，建议先掌握其相关概念再上手代码
- 每一个版本都有着要解决的问题与此版本的最大痛点，带着问题去写代码，并且与上个版本的代码进行比较差异
- RPC的基本概念，核心功能
常见的RPC框架
- 远程通讯
- 基于接口方法的透明远程过程调用
- 负载均衡
- 服务注册中心
client 调用远程方法-> request序列化 -> 协议编码 -> 网络传输-> 服务端 -> 反序列化request -> 调用本地方法得到response -> 序列化 ->编码->…..
从0开始的RPC的迭代过程：
- version0版本：以不到百行的代码完成一个RPC例子
- version1版本：完善通用消息格式（request，response），客户端的动态代理完成对request消息格式的封装
- version2版本：支持服务端暴露多个服务接口， 服务端程序抽象化，规范化
- version3版本：使用高性能网络框架netty的实现网络通信，以及客户端代码的重构
- version4版本：自定义消息格式，支持多种序列化方式（java原生， json…）
- version5版本: 服务器注册与发现的实现，zookeeper作为注册中心
- version6版本: 负载均衡的策略的实现
- version7版本: 客户端缓存服务地址列表, zookeeper监听服务提供者状态，更新客户端缓存**（待实现）**
- version8版本： 跨语言的RPC通信（protobuf）（待实现）
- java基础
- java socket编程入门
- 项目使用maven搭建，暂时只引入了lombok包
- 什么是RPC，怎么完成一个RPC?
一个RPC最最最简单的过程是客户端调用服务端的的一个方法, 服务端返回执行方法的返回值给客服端。接下来我会以一个从数据库里取数据的例子来进行一次模拟RPC过程的一个完整流程。
假定有以下这样一个服务：
服务端：
-
有一个User表
- UserServiceImpl 实现了UserService接口
- UserService里暂时只有一个功能: getUserByUserId(Integer id)
客户端：
传一个Id给服务端，服务端查询到User对象返回给客户端
- 首先我们得有User对象，这是客户端与服务端都已知的，客户端需要得到这个pojo对象数据，服务端需要操作这个对象
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User implements Serializable {
// 客户端和服务端共有的
private Integer id;
private String userName;
private Boolean sex;
}- 定义客户端需要调用，服务端需要提供的服务接口
public interface UserService {
// 客户端通过这个接口调用服务端的实现类
User getUserByUserId(Integer id);
}- 服务端需要实现Service接口的功能
public class UserServiceImpl implements UserService {
@Override
public User getUserByUserId(Integer id) {
System.out.println("客户端查询了"+id+"的用户");
// 模拟从数据库中取用户的行为
Random random = new Random();
User user = User.builder().userName(UUID.randomUUID().toString())
.id(id)
.sex(random.nextBoolean()).build();
return user;
}
}- 客户端建立Socket连接，传输Id给服务端，得到返回的User对象
public class RPCClient {
public static void main(String[] args) {
try {
// 建立Socket连接
Socket socket = new Socket("127.0.0.1", 8899);
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
// 传给服务器id
objectOutputStream.writeInt(new Random().nextInt());
objectOutputStream.flush();
// 服务器查询数据，返回对应的对象
User user = (User) objectInputStream.readObject();
System.out.println("服务端返回的User:"+user);
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
System.out.println("客户端启动失败");
}
}
}- 服务端以BIO的方式监听Socket，如有数据，调用对应服务的实现类执行任务，将结果返回给客户端
public class RPCServer {
public static void main(String[] args) {
UserServiceImpl userService = new UserServiceImpl();
try {
ServerSocket serverSocket = new ServerSocket(8899);
System.out.println("服务端启动了");
// BIO的方式监听Socket
while (true){
Socket socket = serverSocket.accept();
// 开启一个线程去处理
new Thread(()->{
try {
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
// 读取客户端传过来的id
Integer id = ois.readInt();
User userByUserId = userService.getUserByUserId(id);
// 写入User对象给客户端
oos.writeObject(userByUserId);
oos.flush();
} catch (IOException e){
e.printStackTrace();
System.out.println("从IO中读取数据错误");
}
}).start();
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("服务器启动失败");
}
}
}
这个例子以不到百行的代码，实现了客户端与服务端的一个远程过程调用，非常适合上手，当然它是及其不完善的，甚至连消息格式都没有统一，我们将在接下来的版本更新中逐渐完善它。
- 只能调用服务端Service唯一确定的方法，如果有两个方法需要调用呢?（Reuest需要抽象）
- 返回值只支持User对象，如果需要传一个字符串或者一个Dog，String对象呢（Response需要抽象）
- 客户端不够通用，host，port， 与调用的方法都特定（需要抽象）
- 反射
- 动态代理
-
如何使客户端请求远程方法支持多种?
-
如何使服务端返回值的类型多样?
更新1：定义了一个通用的Request的对象（消息格式）
/**
* 在上个例子中，我们的Request仅仅只发送了一个id参数过去，这显然是不合理的，
* 因为服务端不会只有一个服务一个方法，因此只传递参数不会知道调用那个方法
* 因此一个RPC请求中，client发送应该是需要调用的Service接口名，方法名，参数，参数类型
* 这样服务端就能根据这些信息根据反射调用相应的方法
* 还是使用java自带的序列化方式
*/
@Data
@Builder
public class RPCRequest implements Serializable {
// 服务类名，客户端只知道接口名，在服务端中用接口名指向实现类
private String interfaceName;
// 方法名
private String methodName;
// 参数列表
private Object[] params;
// 参数类型
private Class<?>[] paramsTypes;
}更新2： 定义了一个通用的Response的对象（消息格式）
/**
* 上个例子中response传输的是User对象，显然在一个应用中我们不可能只传输一种类型的数据
* 由此我们将传输对象抽象成为Object
* Rpc需要经过网络传输，有可能失败，类似于http，引入状态码和状态信息表示服务调用成功还是失败
*/
@Data
@Builder
public class RPCResponse implements Serializable {
// 状态信息
private int code;
private String message;
// 具体数据
private Object data;
public static RPCResponse success(Object data) {
return RPCResponse.builder().code(200).data(data).build();
}
public static RPCResponse fail() {
return RPCResponse.builder().code(500).message("服务器发生错误").build();
}
}因此在网络传输过程都是request与response格式的数据了，客户端与服务器端就要负责封装与解析以上结构数据
更新3： 服务端接受request请求，并调用request中的对应的方法
public interface UserService {
// 客户端通过这个接口调用服务端的实现类
User getUserByUserId(Integer id);
// 给这个服务增加一个功能
Integer insertUserId(User user);
}服务端的实现类UserServiceImpl要实现增加的功能
@Override
public Integer insertUserId(User user) {
System.out.println("插入数据成功："+user);
return 1;
}服务端接受解析reuqest与封装发送response对象
public class RPCServer {
public static void main(String[] args) {
UserServiceImpl userService = new UserServiceImpl();
try {
ServerSocket serverSocket = new ServerSocket(8899);
System.out.println("服务端启动了");
// BIO的方式监听Socket
while (true){
Socket socket = serverSocket.accept();
// 开启一个线程去处理，这个类负责的功能太复杂，以后代码重构中，这部分功能要分离出来
new Thread(()->{
try {
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
// 读取客户端传过来的request
RPCRequest request = (RPCRequest) ois.readObject();
// 反射调用对应方法
Method method = userService.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
Object invoke = method.invoke(userService, request.getParams());
// 封装，写入response对象
oos.writeObject(RPCResponse.success(invoke));
oos.flush();
}catch (IOException | ClassNotFoundException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e){
e.printStackTrace();
System.out.println("从IO中读取数据错误");
}
}).start();
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("服务器启动失败");
}
}
}更新4： 客户端根据不同的Service进行动态代理：
代理对象增强的公共行为：把不同的Service方法封装成统一的Request对象格式，并且建立与Server的通信
- 底层的通信
public class IOClient {
// 这里负责底层与服务端的通信，发送的Request，接受的是Response对象
// 客户端发起一次请求调用，Socket建立连接，发起请求Request，得到响应Response
// 这里的request是封装好的（上层进行封装），不同的service需要进行不同的封装， 客户端只知道Service接口，需要一层动态代理根据反射封装不同的Service
public static RPCResponse sendRequest(String host, int port, RPCRequest request){
try {
Socket socket = new Socket(host, port);
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
System.out.println(request);
objectOutputStream.writeObject(request);
objectOutputStream.flush();
RPCResponse response = (RPCResponse) objectInputStream.readObject();
return response;
} catch (IOException | ClassNotFoundException e) {
System.out.println();
return null;
}
}
}- 动态代理封装request对象
@AllArgsConstructor
public class ClientProxy implements InvocationHandler {
// 传入参数Service接口的class对象，反射封装成一个request
private String host;
private int port;
// jdk 动态代理， 每一次代理对象调用方法，会经过此方法增强（反射获取request对象，socket发送至客户端）
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// request的构建，使用了lombok中的builder，代码简洁
RPCRequest request = RPCRequest.builder().interfaceName(method.getDeclaringClass().getName())
.methodName(method.getName())
.params(args).paramsTypes(method.getParameterTypes()).build();
// 数据传输
RPCResponse response = IOClient.sendRequest(host, port, request);
//System.out.println(response);
return response.getData();
}
<T>T getProxy(Class<T> clazz){
Object o = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, this);
return (T)o;
}
}- 客户端调用不同的方法
public class RPCClient {
public static void main(String[] args) {
ClientProxy clientProxy = new ClientProxy("127.0.0.1", 8899);
UserService proxy = clientProxy.getProxy(UserService.class);
// 服务的方法1
User userByUserId = proxy.getUserByUserId(10);
System.out.println("从服务端得到的user为：" + userByUserId);
// 服务的方法2
User user = User.builder().userName("张三").id(100).sex(true).build();
Integer integer = proxy.insertUserId(user);
System.out.println("向服务端插入数据："+integer);
}
}
- 定义更加通用的消息格式：Request 与Response格式， 从此可能调用不同的方法，与返回各种类型的数据。
- 使用了动态代理进行不同服务方法的Request的封装，
- 客户端更加松耦合，不再与特定的Service，host，port绑定
- 服务端我们直接绑定的是UserService服务，如果还有其它服务接口暴露呢?（多个服务的注册）
- 服务端以BIO的方式性能是否太低，
- 服务端功能太复杂：监听，处理。需要松耦合
- 代码解耦
- 线程池
如果一个服务端要提供多个服务的接口， 例如新增一个BlogService，怎么处理?
// 自然的想到用一个Map来保存，<interfaceName, xxxServiceImpl>
UserService userService = new UserServiceImpl();
BlogService blogService = new BlogServiceImpl();
Map<String, Object>.put("***.userService", userService);
Map<String, Object>.put("***.blogService", blogService);
// 此时来了一个request，我们就能从map中取出对应的服务
Object service = map.get(request.getInterfaceName())更新前的工作： 更新一个新的服务接口样例和pojo类
// 新的服务接口
public interface BlogService {
Blog getBlogById(Integer id);
}
// 服务端新的服务接口实现类
public class BlogServiceImpl implements BlogService {
@Override
public Blog getBlogById(Integer id) {
Blog blog = Blog.builder().id(id).title("我的博客").useId(22).build();
System.out.println("客户端查询了"+id+"博客");
return blog;
}
}
// pojo类
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Blog implements Serializable {
private Integer id;
private Integer useId;
private String title;
}更新1： HashMap<String, Object> 添加多个服务的实现类
public class TestServer {
public static void main(String[] args) {
UserService userService = new UserServiceImpl();
BlogService blogService = new BlogServiceImpl();
Map<String, Object> serviceProvide = new HashMap<>();
// 暴露两个服务接口， 即在RPCServer中加一个HashMap
serviceProvide.put("com.ganghuan.myRPCVersion2.service.UserService",userService);
serviceProvide.put("com.ganghuan.myRPCVersion2.service.BlogService",blogService);
RPCServer RPCServer = new SimpleRPCRPCServer(serviceProvide);
RPCServer.start(8899);
}
}
// 这里先不去讨论实现其中的细节，因为这里还应该进行优化，我们先去把服务端代码松耦合，再回过来讨论更新2： 服务端代码重构
- 抽象RPCServer，开放封闭原则
// 把RPCServer抽象成接口，以后的服务端实现这个接口即可
public interface RPCServer {
void start(int port);
void stop();
}- RPCService简单版本的实现
/**
* 这个实现类代表着java原始的BIO监听模式，来一个任务，就new一个线程去处理
* 处理任务的工作见WorkThread中
*/
public class SimpleRPCRPCServer implements RPCServer {
// 存着服务接口名-> service对象的map
private Map<String, Object> serviceProvide;
public SimpleRPCRPCServer(Map<String,Object> serviceProvide){
this.serviceProvide = serviceProvide;
}
public void start(int port) {
try {
ServerSocket serverSocket = new ServerSocket(port);
System.out.println("服务端启动了");
// BIO的方式监听Socket
while (true){
Socket socket = serverSocket.accept();
// 开启一个新线程去处理
new Thread(new WorkThread(socket,serviceProvide)).start();
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("服务器启动失败");
}
}
public void stop(){
}
}- 为了加强性能，我们还提供了线程池版的实现
public class ThreadPoolRPCRPCServer implements RPCServer {
private final ThreadPoolExecutor threadPool;
private Map<String, Object> serviceProvide;
public ThreadPoolRPCRPCServer(Map<String, Object> serviceProvide){
threadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
1000, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
this.serviceProvide = serviceProvide;
}
public ThreadPoolRPCRPCServer(Map<String, Object> serviceProvide, int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue){
threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
this.serviceProvide = serviceProvide;
}
@Override
public void start(int port) {
System.out.println("服务端启动了");
try {
ServerSocket serverSocket = new ServerSocket(port);
while(true){
Socket socket = serverSocket.accept();
threadPool.execute(new WorkThread(socket,serviceProvide));
}
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void stop() {
}
}- 工作任务类，从服务端代码分离出来，简化服务端代码，单一职责原则
/**
* 这里负责解析得到的request请求，执行服务方法，返回给客户端
* 1. 从request得到interfaceName 2. 根据interfaceName在serviceProvide Map中获取服务端的实现类
* 3. 从request中得到方法名，参数， 利用反射执行服务中的方法 4. 得到结果，封装成response，写入socket
*/
@AllArgsConstructor
public class WorkThread implements Runnable{
private Socket socket;
private Map<String, Object> serviceProvide;
@Override
public void run() {
try {
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
// 读取客户端传过来的request
RPCRequest request = (RPCRequest) ois.readObject();
// 反射调用服务方法获得返回值
RPCResponse response = getResponse(request);
//写入到客户端
oos.writeObject(response);
oos.flush();
}catch (IOException | ClassNotFoundException e){
e.printStackTrace();
System.out.println("从IO中读取数据错误");
}
}
private RPCResponse getResponse(RPCRequest request){
// 得到服务名
String interfaceName = request.getInterfaceName();
// 得到服务端相应服务实现类
Object service = serviceProvide.get(interfaceName);
// 反射调用方法
Method method = null;
try {
method = service.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
Object invoke = method.invoke(service, request.getParams());
return RPCResponse.success(invoke);
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
e.printStackTrace();
System.out.println("方法执行错误");
return RPCResponse.fail();
}
}
}服务端代码第一次重构完毕。
更新3： 服务暴露类，这里回到了更新1**，我们发现服务接口名是我们**直接手写的，这里其实可以利用class对象自动得到
/**
* 之前这里使用Map简单实现的
* 存放服务接口名与服务端对应的实现类
* 服务启动时要暴露其相关的实现类0
* 根据request中的interface调用服务端中相关实现类
*/
public class ServiceProvider {
/**
* 一个实现类可能实现多个接口
*/
private Map<String, Object> interfaceProvider;
public ServiceProvider(){
this.interfaceProvider = new HashMap<>();
}
public void provideServiceInterface(Object service){
String serviceName = service.getClass().getName();
Class<?>[] interfaces = service.getClass().getInterfaces();
for(Class clazz : interfaces){
interfaceProvider.put(clazz.getName(),service);
}
}
public Object getService(String interfaceName){
return interfaceProvider.get(interfaceName);
}
}前面服务端的代码中有Sevicerprovide 这个HashMap的地方需要改成ServiProvider，比如
public class TestServer {
public static void main(String[] args) {
UserService userService = new UserServiceImpl();
BlogService blogService = new BlogServiceImpl();
// Map<String, Object> serviceProvide = new HashMap<>();
// serviceProvide.put("com.ganghuan.myRPCVersion2.service.UserService",userService);
// serviceProvide.put("com.ganghuan.myRPCVersion2.service.BlogService",blogService);
ServiceProvider serviceProvider = new ServiceProvider();
serviceProvider.provideServiceInterface(userService);
serviceProvider.provideServiceInterface(blogService);
RPCServer RPCServer = new SimpleRPCRPCServer(serviceProvider);
RPCServer.start(8899);
}
}// 客户中添加新的测试用例
BlogService blogService = rpcClientProxy.getProxy(BlogService.class);
Blog blogById = blogService.getBlogById(10000);
System.out.println("从服务端得到的blog为：" + blogById);
在一版本中，我们重构了服务端的代码，代码更加简洁，
添加线程池版的服务端的实现，性能应该会有所提升（未测）
并且服务端终于能够提供不同服务了， 功能更加完善，不再鸡肋
- 传统的BIO与线程池网络传输性能低
- netty高性能网络框架的使用
如何提升这个rpc的性能? 可以从两个方面入手，网络传输从BIO到NIO，序列化要减少字节流长度，提高序列化反序列化效率
知名的rpc框架：dubbo， grpc都是使用netty底层进行通信的
前提： maven pom.xml文件引入netty
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.51.Final</version>
</dependency>升级1： 重构客户端代码
客户端的代码太乱了， 我们先进行代码重构，才有利于后面使用netty的方式实现客户端，使之不同方式网络连接的客户端有着同样的结构，同样的api
假如我们现在已经有了两个客户端：SimpleRPCClient(使用java BIO的方式)， NettyRPCClient（使用netty进行网络传输），那么它们两者的共性是啥?发送请求与得到response是共性， 而建立连接与发送请求的方式是不同点。
// 共性抽取出来
public interface RPCClient {
RPCResponse sendRequest(RPCRequest response);
}
// SimpleRPCClient实现这个接口，不同的网络方式有着不同的实现
@AllArgsConstructor
public class SimpleRPCClient implements RPCClient{
private String host;
private int port;
// 客户端发起一次请求调用，Socket建立连接，发起请求Request，得到响应Response
// 这里的request是封装好的，不同的service需要进行不同的封装， 客户端只知道Service接口，需要一层动态代理根据反射封装不同的Service
public RPCResponse sendRequest(RPCRequest request) {
try {
// 发起一次Socket连接请求
Socket socket = new Socket(host, port);
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
System.out.println(request);
objectOutputStream.writeObject(request);
objectOutputStream.flush();
RPCResponse response = (RPCResponse) objectInputStream.readObject();
//System.out.println(response.getData());
return response;
} catch (IOException | ClassNotFoundException e) {
System.out.println();
return null;
}
}
}而RPCClientProxy类中需要加入一个RPCClient类变量即可， 传入不同的client(simple,netty), 即可调用公共的接口sendRequest发送请求，所以客户端代码结构很清晰了:
-
RPCClient: 不同的网络连接，网络传输方式的客户端分别实现这个接口
-
XXXRPCClient: 具体实现类
-
RPCClientProxy： 动态代理Service类，封装不同的Service请求为Request对象，并且持有一个RPCClient对象，负责与服务端的通信，
由此，客户端代码重构完毕，结构更为清晰，一个使用用例为：
// 构建一个使用java Socket传输的客户端
SimpleRPCClient simpleRPCClient = new SimpleRPCClient("127.0.0.1", 8899);
// 把这个客户端传入代理客户端
RPCClientProxy rpcClientProxy = new RPCClientProxy(simpleRPCClient);
// 代理客户端根据不同的服务，获得一个代理类， 并且这个代理类的方法以或者增强（封装数据，发送请求）
UserService userService = rpcClientProxy.getProxy(UserService.class);
// 调用方法
User userByUserId = userService.getUserByUserId(10);升级2： 使用netty方式传输数据：实现NettyRPCServer， NettyRPCCLient，这里建议先学习下netty的启动代码
netty 服务端的实现
/**
* 实现RPCServer接口，负责监听与发送数据
*/
@AllArgsConstructor
public class NettyRPCServer implements RPCServer {
private ServiceProvider serviceProvider;
@Override
public void start(int port) {
// netty 服务线程组boss负责建立连接， work负责具体的请求
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
System.out.printf("Netty服务端启动了...");
try {
// 启动netty服务器
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 初始化
serverBootstrap.group(bossGroup,workGroup).channel(NioServerSocketChannel.class)
.childHandler(new NettyServerInitializer(serviceProvider));
// 同步阻塞
ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
// 死循环监听
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
@Override
public void stop() {
}
}netty server初始化类
/**
* 初始化，主要负责序列化的编码解码， 需要解决netty的粘包问题
*/
@AllArgsConstructor
public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {
private ServiceProvider serviceProvider;
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 消息格式 [长度][消息体], 解决粘包问题
pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
// 计算当前待发送消息的长度，写入到前4个字节中
pipeline.addLast(new LengthFieldPrepender(4));
// 这里使用的还是java 序列化方式， netty的自带的解码编码支持传输这种结构
pipeline.addLast(new ObjectEncoder());
pipeline.addLast(new ObjectDecoder(new ClassResolver() {
@Override
public Class<?> resolve(String className) throws ClassNotFoundException {
return Class.forName(className);
}
}));
pipeline.addLast(new NettyRPCServerHandler(serviceProvider));
}
}netty server具体的handler
/**
* 因为是服务器端，我们知道接受到请求格式是RPCRequest
* Object类型也行，强制转型就行
*/
@AllArgsConstructor
public class NettyRPCServerHandler extends SimpleChannelInboundHandler<RPCRequest> {
private ServiceProvider serviceProvider;
@Override
protected void channelRead0(ChannelHandlerContext ctx, RPCRequest msg) throws Exception {
//System.out.println(msg);
RPCResponse response = getResponse(msg);
ctx.writeAndFlush(response);
ctx.close();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
RPCResponse getResponse(RPCRequest request) {
// 得到服务名
String interfaceName = request.getInterfaceName();
// 得到服务端相应服务实现类
Object service = serviceProvider.getService(interfaceName);
// 反射调用方法
Method method = null;
try {
method = service.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
Object invoke = method.invoke(service, request.getParams());
return RPCResponse.success(invoke);
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
e.printStackTrace();
System.out.println("方法执行错误");
return RPCResponse.fail();
}
}
}客户端netty的实现
/**
* 实现RPCClient接口
*/
public class NettyRPCClient implements RPCClient {
private static final Bootstrap bootstrap;
private static final EventLoopGroup eventLoopGroup;
private String host;
private int port;
public NettyRPCClient(String host, int port) {
this.host = host;
this.port = port;
}
// netty客户端初始化，重复使用
static {
eventLoopGroup = new NioEventLoopGroup();
bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new NettyClientInitializer());
}
/**
* 这里需要操作一下，因为netty的传输都是异步的，你发送request，会立刻返回， 而不是想要的相应的response
*/
@Override
public RPCResponse sendRequest(RPCRequest request) {
try {
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
Channel channel = channelFuture.channel();
// 发送数据
channel.writeAndFlush(request);
channel.closeFuture().sync();
// 阻塞的获得结果，通过给channel设计别名，获取特定名字下的channel中的内容（这个在hanlder中设置）
// AttributeKey是，线程隔离的，不会由线程安全问题。
// 实际上不应通过阻塞，可通过回调函数
AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
RPCResponse response = channel.attr(key).get();
System.out.println(response);
return response;
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
}初始化： 与服务端一致，就不贴代码了
ClientHandler设计：
public class NettyClientHandler extends SimpleChannelInboundHandler<RPCResponse> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RPCResponse msg) throws Exception {
// 接收到response, 给channel设计别名，让sendRequest里读取response
AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
ctx.channel().attr(key).set(msg);
ctx.channel().close();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
此版本我们完成了客户端的重构，使之能够支持多种版本客户端的扩展（实现RPCClient接口）
并且使用netty实现了客户端与服务端的通信
- java自带序列化方式（Java序列化写入不仅是完整的类名，也包含整个类的定义，包含所有被引用的类），不够通用，不够高效
- 各种序列化方式以及比较，（Java原生序列化， json，protobuf，kryo..）参考博客
- 自定义协议
- java IO 了解
- TCP粘包问题，解决方式
-
如何设计并完成自己的协议?
答：自己实现encode与decode
在前面的RPC版本中， 我们都是使用的java自带的序列化的方式，事实上使用这种方式性能是很低的（序列化后的字节数组， 解码编码速度），而且在netty服务端，我们使用的是netty自带的编码器，简单的传输了一个 消息长度（4个字节）| 序列化后的数据 格式的数据。 在这里我们要自定义我们的传输格式和编解码。
下面是我的初步对我的自定义格式的设计了， 先读取消息类型（Requst， Response， ping， pong）， 序列化方式（原生， json，kryo， protobuf..）， 加上消息长度：防止粘包， 再根据长度读取data
| 消息类型（2Byte） | 序列化方式 2Byte | 消息长度 4Byte |
|---|---|---|
| 序列化后的Data…. | 序列化后的Data… | 序列化后的Data…. |
前提处理： maven pom文件中引入fastjson包， RPCResponse中需要加入DataType字段，因为其它序列化方式（json）无法得到Data的类型，
升级1： 自定义编解码器
序列化器的接口：
public interface Serializer {
// 把对象序列化成字节数组
byte[] serialize(Object obj);
// 从字节数组反序列化成消息, 使用java自带序列化方式不用messageType也能得到相应的对象（序列化字节数组里包含类信息）
// 其它方式需指定消息格式，再根据message转化成相应的对象
Object deserialize(byte[] bytes, int messageType);
// 返回使用的序列器，是哪个
// 0：java自带序列化方式, 1: json序列化方式
int getType();
// 根据序号取出序列化器，暂时有两种实现方式，需要其它方式，实现这个接口即可
static Serializer getSerializerByCode(int code){
switch (code){
case 0:
return new ObjectSerializer();
case 1:
return new JsonSerializer();
default:
return null;
}
}
}encode类
/**
* 依次按照自定义的消息格式写入，传入的数据为request或者response
* 需要持有一个serialize器，负责将传入的对象序列化成字节数组
*/
@AllArgsConstructor
public class MyEncode extends MessageToByteEncoder {
private Serializer serializer;
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
System.out.println(msg.getClass());
// 写入消息类型
if(msg instanceof RPCRequest){
out.writeShort(MessageType.REQUEST.getCode());
}
else if(msg instanceof RPCResponse){
out.writeShort(MessageType.RESPONSE.getCode());
}
// 写入序列化方式
out.writeShort(serializer.getType());
// 得到序列化数组
byte[] serialize = serializer.serialize(msg);
// 写入长度
out.writeInt(serialize.length);
// 写入序列化字节数组
out.writeBytes(serialize);
}
}decode类
/**
* 按照自定义的消息格式解码数据
*/
@AllArgsConstructor
public class MyDecode extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
// 1. 读取消息类型
short messageType = in.readShort();
// 现在还只支持request与response请求
if(messageType != MessageType.REQUEST.getCode() &&
messageType != MessageType.RESPONSE.getCode()){
System.out.println("暂不支持此种数据");
return;
}
// 2. 读取序列化的类型
short serializerType = in.readShort();
// 根据类型得到相应的序列化器
Serializer serializer = Serializer.getSerializerByCode(serializerType);
if(serializer == null)throw new RuntimeException("不存在对应的序列化器");
// 3. 读取数据序列化后的字节长度
int length = in.readInt();
// 4. 读取序列化数组
byte[] bytes = new byte[length];
in.readBytes(bytes);
// 用对应的序列化器解码字节数组
Object deserialize = serializer.deserialize(bytes, messageType);
out.add(deserialize);
}
}升级2： 支持不同的序列化器
Java 原生序列化
public class ObjectSerializer implements Serializer{
// 利用java IO 对象 -> 字节数组
@Override
public byte[] serialize(Object obj) {
byte[] bytes = null;
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try {
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(obj);
oos.flush();
bytes = bos.toByteArray();
oos.close();
bos.close();
} catch (IOException e) {
e.printStackTrace();
}
return bytes;
}
// 字节数组 -> 对象
@Override
public Object deserialize(byte[] bytes, int messageType) {
Object obj = null;
ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
try {
ObjectInputStream ois = new ObjectInputStream(bis);
obj = ois.readObject();
ois.close();
bis.close();
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
}
return obj;
}
// 0 代表java原生序列化器
@Override
public int getType() {
return 0;
}
}Json序列化器
/**
* 由于json序列化的方式是通过把对象转化成字符串，丢失了Data对象的类信息，所以deserialize需要
* 了解对象对象的类信息，根据类信息把JsonObject -> 对应的对象
*/
public class JsonSerializer implements Serializer{
@Override
public byte[] serialize(Object obj) {
byte[] bytes = JSONObject.toJSONBytes(obj);
return bytes;
}
@Override
public Object deserialize(byte[] bytes, int messageType) {
Object obj = null;
// 传输的消息分为request与response
switch (messageType){
case 0:
RPCRequest request = JSON.parseObject(bytes, RPCRequest.class);
Object[] objects = new Object[request.getParams().length];
// 把json字串转化成对应的对象， fastjson可以读出基本数据类型，不用转化
for(int i = 0; i < objects.length; i++){
Class<?> paramsType = request.getParamsTypes()[i];
if (!paramsType.isAssignableFrom(request.getParams()[i].getClass())){
objects[i] = JSONObject.toJavaObject((JSONObject) request.getParams()[i],request.getParamsTypes()[i]);
}else{
objects[i] = request.getParams()[i];
}
}
request.setParams(objects);
obj = request;
break;
case 1:
RPCResponse response = JSON.parseObject(bytes, RPCResponse.class);
Class<?> dataType = response.getDataType();
if(! dataType.isAssignableFrom(response.getData().getClass())){
response.setData(JSONObject.toJavaObject((JSONObject) response.getData(),dataType));
}
obj = response;
break;
default:
System.out.println("暂时不支持此种消息");
throw new RuntimeException();
}
return obj;
}
// 1 代表着json序列化方式
@Override
public int getType() {
return 1;
}
}在netty初始化类（客户端，服务器端）中添加解码器中使用自己定义的解码器：
public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 使用自定义的编解码器
pipeline.addLast(new MyDecode());
// 编码需要传入序列化器，这里是json，还支持ObjectSerializer，也可以自己实现其他的
pipeline.addLast(new MyEncode(new JsonSerializer()));
pipeline.addLast(new NettyClientHandler());
}
}
成功启动!
在这版本中，我们自己定义的消息格式，使之支持多种消息类型，序列化方式，使用消息头加长度的方式解决粘包问题
并且，实现了ObjectSerializer与JsonSerializer两种序列化器，也可以轻松扩展为其它序列化方式（实现Serialize接口）。
- 服务端与客户端通信的host与port预先就必须知道的，每一个客户端都必须知道对应服务的ip与端口号， 并且如果服务挂了或者换地址了，就很麻烦。扩展性也不强
- zookeeper安装， 基本概念
- 了解curator开源zookeeper客户端中的使用
- 如何设计一个注册中心
注册中心（如zookeeper）的地址是固定的（为了高可用一般是集群，我们看做黑盒即可）， 服务端上线时，在注册中心注册自己的服务与对应的地址，而客户端调用服务时，去注册中心根据服务名找到对应的服务端地址。
zookeeper我们可以近似看作一个树形目录文件系统，是一个分布式协调应用，其它注册中心有EureKa， Nacos等
前提
- 下载解压Zookeeper [地址]（https://zookeeper.apache.org/releases.html）
- 学习一个zookeeper启动的例子 官方例子
- zoo.cfg 修改dataDir为一个存在目录
- windows启动命令: bin/zkServer.cmd
- java项目中引入Curator客户端，
<!--这个jar包应该依赖log4j,不引入log4j会有控制台会有warn，但不影响正常使用-->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.1.0</version>
</dependency>更新1 ： 引入zookeeper作为注册中心
启动本地zookeeper服务端，默认端口2181。zookeeper客户端测试如下：
先定义服务注册接口
// 服务注册接口，两大基本功能，注册：保存服务与地址。 查询：根据服务名查找地址
public interface ServiceRegister {
void register(String serviceName, InetSocketAddress serverAddress);
InetSocketAddress serviceDiscovery(String serviceName);
}zookeeper服务注册接口的实现类
public class ZkServiceRegister implements ServiceRegister{
// curator 提供的zookeeper客户端
private CuratorFramework client;
// zookeeper根路径节点
private static final String ROOT_PATH = "MyRPC";
// 这里负责zookeeper客户端的初始化，并与zookeeper服务端建立连接
public ZkServiceRegister(){
// 指数时间重试
RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);
// zookeeper的地址固定，不管是服务提供者还是，消费者都要与之建立连接
// sessionTimeoutMs 与 zoo.cfg中的tickTime 有关系，
// zk还会根据minSessionTimeout与maxSessionTimeout两个参数重新调整最后的超时值。默认分别为tickTime 的2倍和20倍
// 使用心跳监听状态
this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
.sessionTimeoutMs(40000).retryPolicy(policy).namespace(ROOT_PATH).build();
this.client.start();
System.out.println("zookeeper 连接成功");
}
@Override
public void register(String serviceName, InetSocketAddress serverAddress){
try {
// serviceName创建成永久节点，服务提供者下线时，不删服务名，只删地址
if(client.checkExists().forPath("/" + serviceName) == null){
client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/" + serviceName);
}
// 路径地址，一个/代表一个节点
String path = "/" + serviceName +"/"+ getServiceAddress(serverAddress);
// 临时节点，服务器下线就删除节点
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
} catch (Exception e) {
System.out.println("此服务已存在");
}
}
// 根据服务名返回地址
@Override
public InetSocketAddress serviceDiscovery(String serviceName) {
try {
List<String> strings = client.getChildren().forPath("/" + serviceName);
// 这里默认用的第一个，后面加负载均衡
String string = strings.get(0);
return parseAddress(string);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
// 地址 -> XXX.XXX.XXX.XXX:port 字符串
private String getServiceAddress(InetSocketAddress serverAddress) {
return serverAddress.getHostName() +
":" +
serverAddress.getPort();
}
// 字符串解析为地址
private InetSocketAddress parseAddress(String address) {
String[] result = address.split(":");
return new InetSocketAddress(result[0], Integer.parseInt(result[1]));
}
}更新2： 更新客户端得到服务器的方式， 服务端暴露服务时，注册到注册中心
首先new client不需要传入host与name， 而在发送request时，从注册中心获得
// 不需传host，port
RPCClient rpcClient = new NettyRPCClient();客户端的改造
public class SimpleRPCClient implements RPCClient {
private String host;
private int port;
private ServiceRegister serviceRegister;
public SimpleRPCClient() {
// 初始化注册中心，建立连接
this.serviceRegister = new ZkServiceRegister();
}
// 客户端发起一次请求调用，Socket建立连接，发起请求Request，得到响应Response
// 这里的request是封装好的，不同的service需要进行不同的封装， 客户端只知道Service接口，需要一层动态代理根据反射封装不同的Service
public RPCResponse sendRequest(RPCRequest request) {
// 从注册中心获取host，port
InetSocketAddress address = serviceRegister.serviceDiscovery(request.getInterfaceName());
host = address.getHostName();
port = address.getPort();
try {
Socket socket = new Socket(host, port);
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
System.out.println(request);
objectOutputStream.writeObject(request);
objectOutputStream.flush();
RPCResponse response = (RPCResponse) objectInputStream.readObject();
//System.out.println(response.getData());
return response;
} catch (IOException | ClassNotFoundException e) {
System.out.println();
return null;
}
}
}服务端的改造：服务端反而需要把自己的ip，端口给注册中心
ServiceProvider serviceProvider = new ServiceProvider("127.0.0.1", 8899);在服务暴露类加入注册的功能
public class ServiceProvider {
/**
* 一个实现类可能实现多个服务接口，
*/
private Map<String, Object> interfaceProvider;
private ServiceRegister serviceRegister;
private String host;
private int port;
public ServiceProvider(String host, int port){
// 需要传入服务端自身的服务的网络地址
this.host = host;
this.port = port;
this.interfaceProvider = new HashMap<>();
this.serviceRegister = new ZkServiceRegister();
}
public void provideServiceInterface(Object service){
Class<?>[] interfaces = service.getClass().getInterfaces();
for(Class clazz : interfaces){
// 本机的映射表
interfaceProvider.put(clazz.getName(),service);
// 在注册中心注册服务
serviceRegister.register(clazz.getName(),new InetSocketAddress(host,port));
}
}
public Object getService(String interfaceName){
return interfaceProvider.get(interfaceName);
}
}成功运行！
此版本中我们加入了注册中心，终于一个完整的RPC框架三个角色都有了：服务提供者，服务消费者，注册中心
- 根据服务名查询地址时，我们返回的总是第一个IP，导致这个提供者压力巨大，而其它提供者调用不到
我们的RPC总体框架已经有了，接下来就是一些修修补补已经扩展功能的模块了
- 负载均衡
- 如果一个服务有多个提供者支持，如何分散服务提供者的压力
- 负载均衡接口
/**
* 给服务器地址列表，根据不同的负载均衡策略选择一个
*/
public interface LoadBalance {
String balance(List<String> addressList);
}- 两种实现方式
/**
* 随机负载均衡
*/
public class RandomLoadBalance implements LoadBalance{
@Override
public String balance(List<String> addressList) {
Random random = new Random();
int choose = random.nextInt(addressList.size());
System.out.println("负载均衡选择了" + choose + "服务器");
return addressList.get(choose);
}
}/**
* 轮询负载均衡
*/
public class RoundLoadBalance implements LoadBalance{
private int choose = -1;
@Override
public String balance(List<String> addressList) {
choose++;
choose = choose%addressList.size();
return addressList.get(choose);
}
}在服务发现方法中使用负载均衡
List<String> strings = client.getChildren().forPath("/" + serviceName);我们开启两台服务提供者
可以看出负载均衡策略已经成功
这一版本中，我们实现负载均衡的两种策略：随机与轮询。
在这一版本下，一个完整功能的RPC已经出现，当然还有许多性能上与代码质量上的工作需要完成。
- 客户端每次发起请求都要先与zookeeper进行通信得到地址，效率低下。