本项目所有代码可见：https://github.com/weiyu-zeng/SimpleRPC
之前谈到，网络传输使用BIO的方式，我们在simpleRPC-04改为NIO来传输，引入netty的编解码方式。
我们在simpleRPC-04中将使用netty来优化我们的客户端client和服务端server，
创建module：simpleRPC-04
在java下创建package：com.rpc
我们配置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">
<parent>
<artifactId>SimpleRPC</artifactId>
<groupId>org.example</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>simpleRPC-04</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.12</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.51.Final</version>
</dependency>
</dependencies>
</project>
我们的common和simpleRPC-03是一样的：
Blog.java
package com.rpc.common;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Blog implements Serializable {
private Integer id;
private Integer useId;
private String title;
}
RPCRequest.java
package com.rpc.common;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
/**
* 客户端请求的抽象（接口名，方法名，参数，参数类型）
*
*/
@Data
@Builder
public class RPCRequest implements Serializable {
// 服务类名，客户端只知道接口名，在服务端中用接口名指向实现类
private String interfaceName;
// 方法名
private String methodName;
// 参数列表
private Object[] params;
// 参数类型
private Class<?>[] paramsTypes;
}
RPCResponse.java
package com.rpc.common;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
/**
* 服务器端回应的抽象
*/
@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();
}
}
User.java
package com.rpc.common;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* @author zwy
*
* 定义简单User信息
*/
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User implements Serializable {
// 客户端和服务端共有的
private Integer id;
private String userName;
private Boolean sex;
}
同样的，service和simpleRPC-04是一样的：
ServiceProvider.java
package com.rpc.service;
import java.util.HashMap;
import java.util.Map;
/**
* @author zwy
*/
public class ServiceProvider {
/**
* 一个实现类可能实现多个接口
*/
private Map<String, Object> interfaceProvider;
// 构造函数，初始化一个空的 hashmap 赋给 Map<String, Object> interfaceProvider
public ServiceProvider() {
this.interfaceProvider = new HashMap<>();
}
public void provideServiceInterface(Object service) {
// 反射，.getClass().getInterfaces()得到class的interface，按照interfaces name(key)和object(value)存入map
Class<?>[] interfaces = service.getClass().getInterfaces();
for (Class clazz : interfaces) {
interfaceProvider.put(clazz.getName(), service);
}
}
public Object getService(String interfaceName) {
return interfaceProvider.get(interfaceName); // 通过interface name得到object
}
}
BlogService.java
package com.rpc.service;
import com.rpc.common.Blog;
public interface BlogService {
Blog getBlogById(Integer id);
}
BlogServiceImpl.java
package com.rpc.service;
import com.rpc.common.Blog;
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;
}
}
UserService.java
package com.rpc.service;
import com.rpc.common.User;
/**
* @author zwy
*
* 服务器端提供服务的方法的接口
*/
public interface UserService {
// 客户端通过这个接口调用服务端的实现类
User getUserByUserId(Integer id);
// 给这个服务增加一个功能
Integer insertUserId(User user);
}
UserServiceImpl.java
package com.rpc.service;
import com.rpc.common.User;
/**
* @author zwy
*
* 服务器端提供服务的方法
*/
public class UserServiceImpl implements UserService {
@Override
public User getUserByUserId(Integer id) {
// 模拟从数据库中取用户的行为
User user = User.builder()
.id(id)
.userName("he2121")
.sex(true).build();
System.out.println("客户端查询了" + id + "的用户");
return user;
}
@Override
public Integer insertUserId(User user) {
System.out.println("插入数据成功: " + user);
return 1;
}
}
我们的simpleRPC-04客户端和服务端用netty进行改进，
我们的PRCServer.java接口和simpleRPC-03一样：
RPCServer.java
package com.rpc.server;
/**
* @author weiyu_zeng
*
* RPC服务器端：接受，解析request，封装，发送response
*
*/
public interface RPCServer {
void start(int port);
void stop();
}
我们创建一个NettyRPCServer.java来实现RPCServer接口：
代码和注释如下：
package com.rpc.server;
import com.rpc.service.ServiceProvider;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.AllArgsConstructor;
/**
* @author weiyu_zeng
*
* 实现RPCServer接口，负责监听与发送数据
*
* NioEventLoopGroup：实际上就是一个线程池，里面有可执行的Executor#Runnable,同时继承了Iterable 迭代器
* 每一个 NioEventLoopGroup 中都包含了多个NioEventLoop，而每个 NioEventLoop 又绑定着一个线程。
* 一个 NioEventLoop 可以处理多个 Channel 中的 IO 操作，而其只有一个线程。所以对于这个线程资源的使用，
* 就存在了竞争。此时为每一个 NioEventLoop都绑定了一个多跑复用器 Selector，由 Selector 来决定当前 NioEventLoop
* 的线程为哪些 Channel 服务。
*
* ServerBootstrap：负责初始化netty服务器，并且开始监听端口的socket请求。
* ServerBootstrap用一个ServerSocketChannelFactory 来实例化。ServerSocketChannelFactory 有两种选择，
* 一种是NioServerSocketChannelFactory，一种是OioServerSocketChannelFactory。前者使用NIO，后则使用普通的阻塞式IO。
* 它们都需要两个线程池实例作为参数来初始化，一个是boss线程池，一个是worker线程池。
*
* ServerBootstrap.bind(int)：负责绑定端口，当这个方法执行后，ServerBootstrap就可以接受指定端口上的socket连接了。
* 一个ServerBootstrap可以绑定多个端口。bind方法会创建一个serverchannel，并且会将当前的channel注册到eventloop上面.
*
* ChannelFuture：最顶层是继承的jdk 的Future 接口，Future 类就是代表了异步计算的结果。
* Netty 里面的IO操作全部是异步的。这意味着，IO操作会立即返回，但是在调用结束时，无法保证IO操作已完成。取而代之，
* 将会返回给你一个ChannelFuture 实例，提供IO操作的结果信息或状态。
* channelFuture.channel()：返回ChannelFuture关联的Channel；
* channelFuture.channel().closeFuture().sync()相当于在这里阻塞，直到serverchannel关闭。
*
* shutdownGracefully()：所有现有channel将自动关闭,并且应拒绝重新连接尝试
*/
@AllArgsConstructor
public class NettyRPCServer implements RPCServer {
private ServiceProvider serviceProvider;
@Override
public void start(int port) {
// netty服务线程组负责建立连接(TCP/IP连接)，work负责具体的请求
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
System.out.println("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() {
}
}
NettyServerInitializer.java
package com.rpc.server;
import com.rpc.service.ServiceProvider;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolver;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import lombok.AllArgsConstructor;
/**
* @author weiyu_zeng
*
* 初始化，主要负责序列化的编码解码， 需要解决netty的粘包问题
*
* ChannelInitializer；它提供了一个简单的方法来初始化一个 Channel，一种特殊的ChannelInboundHandler，
* 用于在某个Channel注册到EventLoop后，对这个Channel执行一些初始化操作。
* ChannelPipeline：是ChannelHandler的容器，它负责ChannelHandler的管理和事件拦截与调度。
* 内部维护了一个ChannelHandler的链表和迭代器，可以方便地实现ChannelHandler查找、添加、替换和删除
*
* Netty的消息传递都是基于流，通过Channel和Buffer传递的，自然，Object也需要转换成Channel和Buffer来传递
* Netty本身已经给我们写好了这样的转换工具。ObjectEncoder和ObjectDecoder：
* Netty给我们处理自己业务的空间是在灵活的可子定义的Handler上的，也就是说，如果我们自己去做这个转换工作，
* 那么也应该在Handler里去做。而Netty，提供给我们的ObjectEncoder和Decoder也恰恰是一组Handler
*
* LengthFieldBasedFrameDecoder：自定义长度帧解码器，解码器自定义长度解决TCP粘包黏包问题。（
* TCP粘包是指发送方发送的若干个数据包到接收方时粘成一个包。从接收缓冲区来看，后一个包数据的头紧接着前一个数据的尾
* 当TCP连接建立后，Client发送多个报文给Server，TCP协议保证数据可靠性，但无法保证Client发了n个包，服务端也按照n个包接收。
* Client端发送n个数据包，Server端可能收到n-1或n+1个包。
* ）
* 如何解决粘包现象
* 1. 添加特殊符号，接收方通过这个特殊符号将接收到的数据包拆分开 - DelimiterBasedFrameDecoder特殊分隔符解码器
* 2. 每次发送固定长度的数据包 - FixedLengthFrameDecoder定长编码器
* 3. 在消息头中定义长度字段，来标识消息的总长度 - LengthFieldBasedFrameDecoder自定义长度解码器 √
*/
@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));
}
}
NettyRPCServerHandler.java
package com.rpc.server;
import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import com.rpc.service.ServiceProvider;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.AllArgsConstructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
/**
* @author weiyu_zeng
*
* ChannelHandlerContext.writeAndFlush：方法会将数据写到ChannelPipeline中当前ChannelHandler的下一个ChannelHandler开始处理。
*/
@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();
}
// 这里和WorkThread里的getResponse差不多
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();
}
}
}
定义服务端：TestServer.java
package com.rpc.server;
import com.rpc.service.*;
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); // 把userService存入 serviceProvider
serviceProvider.provideServiceInterface(blogService); // 把blogService存入 serviceProvider
// RPCServer RPCServer = new ThreadPoolRPCRPCServer(serviceProvider);
RPCServer RPCServer = new NettyRPCServer(serviceProvider);
RPCServer.start(8899);
}
}
接下来是客户端的改造
RPCClient.java和simpleRPC-03一样：
package com.rpc.client;
import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
/**
* @author weiyu_zeng
*
* RPC客户端：发送请求，获得response
*/
public interface RPCClient {
RPCResponse sendRequest(RPCRequest request);
}
RPCClientProxy.java 代理类需要稍微修改：
package com.rpc.client;
import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import lombok.AllArgsConstructor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
/**
* @author weiyu_zeng
*
* 客户端代理：把动态代理封装request对象（这里和simpleRPC-02的ClientProxy函数一样，保留了动态代理的设计）
*/
@AllArgsConstructor
public class RPCClientProxy implements InvocationHandler {
private RPCClient client;
// 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 = client.sendRequest(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;
}
}
NettyRPCClient.java
package com.rpc.client;
import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
/**
* @author weiyu_zeng
*
* 实现RPCClient接口
*/
public class NettyRPCClient implements RPCClient{
private static final Bootstrap bootstrap;
private static final EventLoopGroup evenLoopGroup;
private String host;
private int port;
// 构造函数
public NettyRPCClient(String host, int port) {
this.host = host;
this.port = port;
}
// netty客户端初始化，重复使用
static {
evenLoopGroup = new NioEventLoopGroup();
bootstrap = new Bootstrap();
bootstrap.group(evenLoopGroup).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;
}
}
NettyClientInitializer.java
package com.rpc.client;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolver;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
/**
* @author weiyu_zeng
*
* 同样的与服务端解码和编码格式
*/
public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
@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));
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 NettyClientHandler());
}
}
NettyClientHandler.java
package com.rpc.client;
import com.rpc.common.RPCResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
/**
* @author weiyu_zeng
*
* AttributeMap这是是绑定在Channel或者ChannelHandlerContext上的一个附件，ChannelHandlerContext中的AttributeMap是独有的，
* Channel上的AttributeMap就是大家共享的，每一个ChannelHandler都能获取到。AttributeMap的结构，其实和Map的格式很像，
* key是AttributeKey，value是Attribute，我们可以根据AttributeKey找到对应的Attribute
*/
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();
}
// 跟NettyRPCServerHandler一样
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
最后定义客户端：
TestClient.java
package com.rpc.client;
import com.rpc.common.Blog;
import com.rpc.common.User;
import com.rpc.service.BlogService;
import com.rpc.service.UserService;
/**
* @author weiyu_zeng
*
* 用例测试的类，继承了很大一部分RPCClient 的main函数。
* 但也新增了新的使用方法。
*/
public class TestClient {
public static void main(String[] args) {
// 构建一个使用java socket或者netty的客户端
RPCClient rpcClient = new NettyRPCClient("127.0.0.1", 8899);
// 把这个客户端传入代理客户端
RPCClientProxy rpcClientProxy = new RPCClientProxy(rpcClient);
// 代理客户端根据不同的服务，获得一个代理类， 并且这个代理类的方法以或者增强（封装数据，发送请求）
UserService userService = rpcClientProxy.getProxy(UserService.class);
// 服务的方法1
User userByUserId = userService.getUserByUserId(10);
System.out.println("从服务器端得到的user为：" + userByUserId);
// 服务的方法2
User user = User.builder().userName("张三").id(100).sex(true).build();
Integer integer = userService.insertUserId(user);
System.out.println("向服务器端插入数据" + integer);
// 服务的方法3
BlogService blogService = rpcClientProxy.getProxy(BlogService.class);
Blog blogById = blogService.getBlogById(10000);
System.out.println("从服务端得到的blog为：" + blogById);
}
}
simpleRPC-04的文件结构如下
我们先运行 TestServer.java
然后运行 TestClient.java
