手撕RPC
使用Netty+Zookeeper+Spring实现简易的RPC框架。阅读本文需要有一些Netty使用基础。
服务信息在网络传输，需要讲服务类进行序列化，服务端使用Spring作为容器。服务端发布服务，将接口的全路径当做节点名称，服务的ip+端口作为节点值，存储到Zookeeper中。客户端调用的时候，去Zookeeper查询，获得提供该接口的服务器ip和端口，通过Netty进行调用。
工程引用的jar包
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.32.Final</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>5.1.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
</dependency>
</dependencies>
1.请求和响应
服务端和客户端进行通信，完成远程服务调用，需要统一封装请求和响应。
请求统一封装类，服务端接收请求之后，通过反射获得客户端传送的对象。
package com.test.rpc.pojo;
import java.io.Serializable;
public class Request implements Serializable {
private String serviceName;//调用的服务名字
private String methodName;//调用的方法名
private Class<?>[] parameterTypes;//方法参数的类型
private Object[] args;//方法的参数
//getter/setter略
}
响应统一封装类，服务端返回给客户端的响应
package com.test.rpc.pojo;
import java.io.Serializable;
public class Response implements Serializable {
private Object result;//响应结果
private Exception exception;//响应的异常
//getter/setter略
}
2.通用工具类
因为对象在网络上传输需要转换成字节，所以需要序列化。
序列化工具类，可以替换成第三方，如json。
package com.test.rpc.util;
import java.io.*;
/**
* 序列化工具，可以替换成第三方
*/
public class SerializeUtil {
//序列化
public static byte[] serialize(Object o) throws IOException {
ByteArrayOutputStream baos = null;
ObjectOutputStream oos = null;
try {
// 序列化
baos = new ByteArrayOutputStream();
oos = new ObjectOutputStream(baos);
oos.writeObject(o);
byte[] bytes = baos.toByteArray();
return bytes;
} catch (Exception e) {
e.printStackTrace();
} finally {
oos.close();
baos.close();
}
return null;
}
// 反序列化
public static Object unserialize(byte[] bytes)throws IOException {
ByteArrayInputStream bais = null;
ObjectInputStream ois = null;
try {
bais = new ByteArrayInputStream(bytes);
ois = new ObjectInputStream(bais);
return ois.readObject();
} catch (Exception e) {
e.printStackTrace();
} finally {
bais.close();
ois.close();
}
return null;
}
}
Zookeeper工具类，连接zookeeper的工具
package com.test.rpc.util;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
public class ZkUtil {
private ZooKeeper zk;
public ZkUtil(String address, int sessionTimeout) {
try {
zk = new ZooKeeper(address, sessionTimeout, null);
} catch (IOException e) {
e.printStackTrace();
}
}
public Stat exist(String path, boolean needWatch) {
try {
return this.zk.exists(path, needWatch);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
public boolean createNode(String path, String data) {
try {
this.exist(path, true);
zk.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
public boolean updateNode(String path, String data) {
try {
this.exist(path, true);
zk.setData(path, data.getBytes(), -1);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
public boolean deleteNode(String path) {
try {
this.exist(path, true);
zk.delete(path, -1);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
public String getNodeData(String path) {
try {
byte[] data = zk.getData(path, false, null);
return new String(data);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
public void close() {
try {
if (zk != null) {
zk.close();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
3.Netty编码解码
package com.test.rpc.netty;
import com.test.rpc.util.SerializeUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
/**
* 消息编码器，实现服务消息序列化
*/
public class ServiceEncoder extends MessageToByteEncoder {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf out) throws Exception {
byte[] bytes = SerializeUtil.serialize(o);//序列化
int dataLength = bytes.length; //读取消息的长度
out.writeInt(dataLength); //先将消息长度写入，也就是消息头,解决粘包和半包的问题
out.writeBytes(bytes);
}
}
package com.test.rpc.netty;
import com.test.rpc.util.SerializeUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;
import java.util.List;
/**
* 通信解码，实现消息的反序列化
*/
public class ServiceDecoder extends ReplayingDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
int length = in.readInt();//消息头，消息的总长度
byte[] content = new byte[length];
in.readBytes(content);
Object o = SerializeUtil.unserialize(content); //将byte数据转化为我们需要的对象。
out.add(o);
}
}
4.服务端代码
为了发布服务方便，定义一个注解，让Spring扫描此注解，统一发布服务
package com.test.rpc.server;
import org.springframework.stereotype.Component;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface RpcService {
Class<?> value();//发布的服务类型
}
注册服务就是把服务名称和提供者地址存到Zookeeper中。
package com.test.rpc.server;
import com.test.rpc.util.ZkUtil;
import org.apache.zookeeper.data.Stat;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import java.util.Map;
public class ServerContext {
private static String ip = "127.0.0.1";//服务端地址
private static int port = 8000;//服务端口
private static final ApplicationContext applicationContext;//spring容器
public static Object getBean(Class clazz) {//获得容器中对象
return applicationContext.getBean(clazz);
}
static {
//创建Spring容器，如果在web环境的话可以用Listener
applicationContext = new ClassPathXmlApplicationContext("spring-server.xml");
ZkUtil zookeeper = applicationContext.getBean(ZkUtil.class);
//通过注解获得bean
Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(RpcService.class);
if (serviceBeanMap != null) {
for (Object bean : serviceBeanMap.values()) {
RpcService rpcService = bean.getClass().getAnnotation(RpcService.class);
String serviceName = "/" + rpcService.value().getName();//注册的节点名,是接口全路径
Stat stat= zookeeper.exist(serviceName,false);
if(stat!=null){//如果节点存在就删除
zookeeper.deleteNode(serviceName);
}
zookeeper.createNode(serviceName, ip + ":" + port);//注册服务
}
}
}
}
package com.test.rpc.server;
import com.test.rpc.pojo.Request;
import com.test.rpc.pojo.Response;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.lang.reflect.Method;
/**
* 处理客户端发送的请求
*/
public class ServerRequestHandler extends SimpleChannelInboundHandler<Request> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Request request) throws Exception {
Response response = new Response();
try {
Object result = handleRequest(request);
response.setResult(result);
} catch (Exception e) {
e.printStackTrace();
response.setException(e);
}
//写入响应并关闭
channelHandlerContext.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
private Object handleRequest(Request request) throws Exception {
String serviceName = request.getServiceName();//被的调用的服务
String methodName = request.getMethodName();//被调用的方法
Class[] parameterTypes = request.getParameterTypes();//方法参数的类型
Object[] args = request.getArgs();//方法的参数
//实例化被调用的服务对象
Class clazz = Class.forName(serviceName);
Object target = ServerContext.getBean(clazz);//从容器获得对象
//获得方法对象
Method method = clazz.getMethod(methodName, parameterTypes);
Object result = method.invoke(target, args);//执行方法
return result;//返回结果
}
}
服务启动类
package com.test.rpc.server;
import com.test.rpc.netty.ServiceDecoder;
import com.test.rpc.netty.ServiceEncoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class ServerMain {
public static void main(String[] args) {
int port=8000;
new ServerContext();
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ServiceEncoder());//编码
ch.pipeline().addLast(new ServiceDecoder());//解码
ch.pipeline().addLast(new ServerRequestHandler());//请求处理
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
5. 待发布的服务
package com.test.rpc.service;
public interface HelloService {
String hello(String message);
}
package com.test.rpc.service;
import com.test.rpc.server.RpcService;
@RpcService(HelloService.class)//自定义的注解
public class HelloServiceImpl implements HelloService {
@Override
public String hello(String message) {
return "server response:" + message;
}
}
spring-server.xml配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="com.test.rpc.service">
</context:component-scan>
<bean id="zkUtil" class="com.test.rpc.util.ZkUtil">
<constructor-arg name="address" value="10.199.0.112"/>
<constructor-arg name="sessionTimeout" value="2000"/>
</bean>
</beans>
6. 客户端代码
package com.test.rpc.client;
import com.test.rpc.netty.ServiceDecoder;
import com.test.rpc.netty.ServiceEncoder;
import com.test.rpc.pojo.Request;
import com.test.rpc.pojo.Response;
import com.test.rpc.util.ZkUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.springframework.context.ApplicationContext;
public class ClientSender extends SimpleChannelInboundHandler<Response> {
private ApplicationContext applicationContext;
public ClientSender(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
private Response response;
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Response response) throws Exception {
this.response = response;//处理响应
}
//发送请求
public Response send(Request request) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
ch.pipeline().addLast(new ServiceEncoder());
ch.pipeline().addLast(new ServiceDecoder());
ch.pipeline().addLast(ClientSender.this);
}
});
//去注册中心查看服务发布者
ZkUtil zk = applicationContext.getBean(ZkUtil.class);
String address = zk.getNodeData("/"+request.getServiceName());
String[] serverAndPort = address.split(":");//查到的服务提供者地址
// 连接 RPC 服务器
ChannelFuture future = bootstrap.connect(serverAndPort[0], Integer.parseInt(serverAndPort[1])).sync();
// 写入 RPC 请求数据并关闭连接
Channel channel = future.channel();
channel.writeAndFlush(request).sync();
channel.closeFuture().sync();
return response;
} finally {
group.shutdownGracefully();
}
}
}
请求动态代理类，调用服务的时候需要创建动态代理，把服务对象封装成Request
package com.test.rpc.client;
import com.test.rpc.pojo.Request;
import com.test.rpc.pojo.Response;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
public class RequestProxy implements ApplicationContextAware {
private ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
//动态代理
public <T> T createProxy(final Class<?> interfaceClass) throws Exception {
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[]{interfaceClass}, new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Request request = new Request();
request.setServiceName(method.getDeclaringClass().getName());
request.setMethodName(method.getName());
request.setParameterTypes(method.getParameterTypes());
request.setArgs(args);
ClientSender sender = new ClientSender(applicationContext);
Response response = sender.send(request);
if (response.getException() != null) {
throw response.getException();
}
return response.getResult();
}
});
}
}
spring-client.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="zkUtil" class="com.test.rpc.util.ZkUtil">
<constructor-arg name="address" value="10.199.0.112"/>
<constructor-arg name="sessionTimeout" value="2000"/>
</bean>
<bean class="com.test.rpc.client.RequestProxy">
</bean>
</beans>
package com.test.rpc.client;
import com.test.rpc.service.HelloService;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ClientMain {
public static void main(String[] args) throws Exception {
ApplicationContext applicationContext=new ClassPathXmlApplicationContext("spring-client.xml");
RequestProxy proxy=applicationContext.getBean(RequestProxy.class);
HelloService helloService=proxy.createProxy(HelloService.class);
System.out.println(helloService.hello("netty"));
}
}