What is RPC
RPC (Remote Procedure Call Protocol), a remote procedure call, has the common explanation that a client calls an object that exists on a remote computer without knowing the details of the call, just like an object in a local application, without knowing the protocol of the underlying network technology.
Simple overall workflow
The requester sends a packet of calls containing parameters required by the protocol such as call identification, parameters, and so on.When the responder receives the packet, the corresponding program is called up, and then the result packet is returned, which contains the same request identification, result, and so on as in the requested packet.
Factors Influencing Performance
- Utilized network protocols.You can use application-layer protocols, such as HTTP or HTTP/2, or transport-layer protocols, such as TCP, but the dominant RPC s do not use UDP transport protocols.
- Message encapsulation format.Select or design a protocol to encapsulate information for assembly and delivery.For example, message body data in dubbo contains dubbo version number, interface name, interface version, method name, parameter type list, parameter, additional information, and so on.
- Serialization.Information is transmitted in binary format in network transmission.Serialization and deserialization are object-to-binary data conversions.Common serialization methods are JSON, Hessian, Protostuff, and so on.
- Network IO model.Non-blocking synchronous IO can be used or multi-channel IO model can be supported on the server.
- Thread management.With high concurrent requests, a single thread can be used to run a specific implementation of the service, but request blocking waits occur.A separate thread can also be opened for each RPC specific service implementation, with a limited maximum number of threads, and a thread pool can be used to manage the allocation and scheduling of multiple threads.
First Edition RPC
The first version simply implements the basic functions of RPC, such as sending and receiving service information, serialization method and dynamic proxy.
Projects use Springboot to implement dependency injection and parameter configuration, netty for NIO-style data transfer, and Hessian for object serialization.
Dynamic Proxy
The proxy mode is mentioned here. It is characterized by the same interface between the proxy class and the delegate class. The proxy class is mainly responsible for pre-processing messages, filtering messages, forwarding messages to the delegate class, and post-processing messages for the delegate class.There is usually an association between a proxy class and a delegate class.
Depending on when the proxy class was created, it can also be divided into static proxy and dynamic proxy.
In previous static proxies, you would need to manually write corresponding proxy classes for each target.If the system already has hundreds or thousands of classes, the workload is too heavy.
Static proxies are created by programmers or by specific tools that automatically generate source code, that is, interfaces and proxy classes, proxy classes, and so on, have been determined at compile time.The.Class file of the proxy class is generated before the program runs.
The way a proxy class creates a proxy when a program is running is called a proxy mode.In a static proxy, the proxy class is self-defined and compiled before running.In dynamic proxies, it is easy to unify the functions of proxy classes without modifying the methods in each proxy class.This can be achieved through the InvocationHandler interface.
Dynamic Proxy for Client
public class ProxyFactory {
public static <T> T create(Class<T> interfaceClass) throws Exception {
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[]{interfaceClass}, new LwRpcClientDynamicProxy<T>(interfaceClass));
}
}@Slf4j
public class LwRpcClientDynamicProxy<T> implements InvocationHandler {
private Class<T> clazz;
public LwRpcClientDynamicProxy(Class<T> clazz) throws Exception {
this.clazz = clazz;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
LwRequest lwRequest = new LwRequest();
String requestId = UUID.randomUUID().toString();
String className = method.getDeclaringClass().getName();
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
lwRequest.setRequestId(requestId);
lwRequest.setClassName(className);
lwRequest.setMethodName(methodName);
lwRequest.setParameterTypes(parameterTypes);
lwRequest.setParameters(args);
NettyClient nettyClient = new NettyClient("127.0.0.1", 8888);
log.info("Start Connecting to Server Side:{}", new Date());
LwResponse send = nettyClient.send(lwRequest);
log.info("Results returned after request:{}", send.getResult());
return send.getResult();
}
}On the server side, the class name obtained on the client side is utilized.Information such as parameters is invoked using a reflection mechanism.
Class<?>[] parameterTypes = request.getParameterTypes();
Object[] paramethers = request.getParameters();
// Use CGLIB reflection
FastClass fastClass = FastClass.create(serviceClass);
FastMethod fastMethod = fastClass.getMethod(methodName, parameterTypes);
return fastMethod.invoke(serviceBean, paramethers);Netty Client
@Slf4j
public class NettyClient {
private String host;
private Integer port;
private LwResponse response;
private EventLoopGroup group;
private ChannelFuture future = null;
private Object obj = new Object();
private NettyClientHandler nettyClientHandler;
public NettyClient(String host, Integer port) {
this.host = host;
this.port = port;
}
public LwResponse send(LwRequest request) throws Exception{
nettyClientHandler = new NettyClientHandler(request);
group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4));
pipeline.addLast(new LwRpcEncoder(LwRequest.class, new HessianSerializer()));
pipeline.addLast(new LwRpcDecoder(LwResponse.class, new HessianSerializer()));
pipeline.addLast(nettyClientHandler);
}
});
future = bootstrap.connect(host, port).sync();
nettyClientHandler.getCountDownLatch().await();
this.response = nettyClientHandler.getLwResponse();
return this.response;
}
@PreDestroy
public void close() {
group.shutdownGracefully();
future.channel().closeFuture().syncUninterruptibly();
}
}@Slf4j
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
private final CountDownLatch countDownLatch = new CountDownLatch(1);
private LwResponse response = null;
private LwRequest request;
public NettyClientHandler(LwRequest request) {
this.request = request;
}
public CountDownLatch getCountDownLatch() {
return countDownLatch;
}
public LwResponse getLwResponse() {
return this.response;
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
log.info("Client sends message to client");
ctx.writeAndFlush(request);
log.info("Client request succeeded");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
LwResponse lwResponse = (LwResponse) msg;
log.info("Receive information from server:{}", lwResponse.getResult());
this.response = lwResponse;
this.countDownLatch.countDown();
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.close();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx,
Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}When the client sends service information, it is encapsulated with the LwQuest class, and the returned result is encapsulated with LwResponse. When the client reads the response returned from the server, it is processed in NettyClientHandler, and the thread is blocked and run with CountDownLatch.
Netty Server
@Component
@Slf4j
public class NettyServer {
private EventLoopGroup boss = null;
private EventLoopGroup worker = null;
@Autowired
private ServerHandler serverHandler;
@Value("${server.address}")
private String address;
public void start() throws Exception {
log.info("Success");
boss = new NioEventLoopGroup();
worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(boss, worker)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4));
pipeline.addLast(new LwRpcEncoder(LwResponse.class, new HessianSerializer()));
pipeline.addLast(new LwRpcDecoder(LwRequest.class, new HessianSerializer()));
pipeline.addLast(serverHandler);
}
});
String[] strs = address.split(":");
String addr = strs[0];
int port = Integer.valueOf(strs[1]);
ChannelFuture future = serverBootstrap.bind(addr, port).sync();
future.channel().closeFuture().sync();
} finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
@PreDestroy
public void destory() throws InterruptedException {
boss.shutdownGracefully().sync();
worker.shutdownGracefully().sync();
log.info("Close netty");
}
}
@Component
@Slf4j
@ChannelHandler.Sharable
public class ServerHandler extends SimpleChannelInboundHandler<LwRequest> implements ApplicationContextAware {
private ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, LwRequest msg) throws Exception {
LwResponse lwResponse = new LwResponse();
lwResponse.setRequestId(msg.getRequestId());
log.info("Receive request information from client:{}", msg);
try {
Object result = handler(msg);
lwResponse.setResult(result);
} catch (Throwable throwable) {
lwResponse.setCause(throwable);
throwable.printStackTrace();
}
channelHandlerContext.writeAndFlush(lwResponse);
}
private Object handler(LwRequest request) throws ClassNotFoundException, InvocationTargetException {
Class<?> clazz = Class.forName(request.getClassName());
Object serviceBean = applicationContext.getBean(clazz);
Class<?> serviceClass = serviceBean.getClass();
String methodName = request.getMethodName();
log.info("Obtained service classes:{}", serviceBean);
Class<?>[] parameterTypes = request.getParameterTypes();
Object[] paramethers = request.getParameters();
// Use CGLIB reflection
FastClass fastClass = FastClass.create(serviceClass);
FastMethod fastMethod = fastClass.getMethod(methodName, parameterTypes);
return fastMethod.invoke(serviceBean, paramethers);
}
}On the Netty server side, the `serverHandler is used to process the information received from the client, invoke local methods with reflected ideas, and encapsulate the structure of the processing in LwResponse.
LwRequest and LwRespnse need to be converted to binary conversion in order to be transmitted over the network.The specific methods are as follows:
public class HessianSerializer implements Serializer {
@Override
public byte[] serialize(Object object) throws IOException {
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
Hessian2Output output = new Hessian2Output(byteArrayOutputStream);
output.writeObject(object);
output.flush();
return byteArrayOutputStream.toByteArray();
}
public <T> T deserialize(Class<T> clazz, byte[] bytes) throws IOException {
Hessian2Input input = new Hessian2Input(new ByteArrayInputStream(bytes));
return (T) input.readObject(clazz);
}
}
public class LwRpcDecoder extends ByteToMessageDecoder {
private Class<?> clazz;
private Serializer serializer;
public LwRpcDecoder(Class<?> clazz, Serializer serializer) {
this.clazz = clazz;
this.serializer = serializer;
}
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
if (byteBuf.readableBytes() < 4)
return;
byteBuf.markReaderIndex();
int dataLength = byteBuf.readInt();
if (dataLength < 0) {
channelHandlerContext.close();
}
if (byteBuf.readableBytes() < dataLength) {
byteBuf.resetReaderIndex();
}
byte[] data = new byte[dataLength];
byteBuf.readBytes(data);
Object obj = serializer.deserialize(clazz, data);
list.add(obj);
}
}
public class LwRpcEncoder extends MessageToByteEncoder<Object> {
private Class<?> clazz;
private Serializer serializer;
public LwRpcEncoder(Class<?> clazz, Serializer serializer) {
this.clazz = clazz;
this.serializer = serializer;
}
@Override
protected void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception {
if (clazz.isInstance(in)) {
byte[] data = serializer.serialize(in);
out.writeInt(data.length);
out.writeBytes(data);
}
}
}