◆ 一.概述
什么是RPC?
那为什么要有 RPC,HTTP 不好么?
项目总体结构
整体架构
接下来,分别解释上述的过程
◆ 二. 自定义注解
服务的提供者和消费者公用一个接口,@ServiceExpose是为了暴露服务,放在生产者的某个实现类上;@ServiceReference是为了引用服务,放在消费者的需要注入的属性上。
◆ 二.启动配置
主要是加载一些rpc相关的配置类,使用SpringBoot自动装配。可以使用SPI机制加入一些自定义的类,放到指定文件夹中。
◆ 三. rpc接口注入/rpc服务扫描
这里主要就是通过反射获得对应注解的属性/类,进行服务暴露/服务引用。这里需要关注的是什么时候进行服务暴露/引用?如下:
那么怎么知道Spring IOC刷新完成,这里就使用一个Spring提供的监听器,当Spring IOC刷新完成,就会触发监听器。
◆ 四. 服务注册到ZK/从Zk获得服务
Zookeeper采用节点树的数据模型,类似linux文件系统,/,/node1,/node2 比较简单。不懂Zookeeper请移步:Zookeeper原理
我们采用的是对每个服务名创建一个持久节点,服务注册时实际上就是在zookeeper中该持久节点下创建了一个临时节点,该临时节点存储了服务的IP、端口、序列化方式等。
客户端获取服务时通过获取持久节点下的临时节点列表,解析服务地址数据:
客户端监听服务变化:
◆ 五. 生成代理类对象
这里使用JDK的动态代理,也可以使用cglib或者Javassist(dobbo使用)。
public class ClientProxyFactory {
/**
* 获取代理对象,绑定 invoke 行为
*
* @param clazz 接口 class 对象
* @param <T> 类型
* @return 代理对象
*/public <T> T getProxyInstance(Class<T> clazz) {
return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new InvocationHandler() {
final Random random = new Random();
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 第一步:通过服务发现机制选择一个服务提供者暴露的服务
String serviceName = clazz.getName();
final List<ServiceInfo> serviceInfos = serviceDiscovery.listServices(serviceName);
logger.info("Rpc server instance list: {}", serviceInfos);
if (CollectionUtils.isEmpty(serviceInfos)) {
throw new RpcException("No rpc servers found.");
}
// TODO: 这里模拟负载均衡,从多个服务提供者暴露的服务中随机挑选一个,后期写方法实现负载均衡
final ServiceInfo serviceInfo = serviceInfos.get(random.nextInt(serviceInfos.size()));
// 第二步:构造 rpc 请求对象
final RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setServiceName(serviceName);
rpcRequest.setMethod(method.getName());
rpcRequest.setParameterTypes(method.getParameterTypes());
rpcRequest.setParameters(args);
// 第三步:编码请求消息, TODO: 这里可以配置多种编码方式
byte[] data = messageProtocol.marshallingReqMessage(rpcRequest);
// 第四步:调用 rpc client 开始发送消息
byte[] byteResponse = rpcClient.sendMessage(data, serviceInfo);
// 第五步:解码响应消息
final RpcResponse rpcResponse = messageProtocol.unmarshallingRespMessage(byteResponse);
// 第六步:解析返回结果进行处理
if (rpcResponse.getException() != null) {
throw rpcResponse.getException();
}
return rpcResponse.getRetValue();
}
});
}
}
◆ 六.负载均衡
本实现支持两种主要负载均衡策略,随机和轮询,其中他们都支持带权重的随机和轮询,其实也就是四种策略。
◆ 七.Netty通信
服务端和客户端基本一样,这里只展示服务端的代码。代理对象在Spring启动的时候就生成了,但是没有调用,每一个调用(请求)都会生成一个Netty的连接。
public class NettyRpcServer extends RpcServer {
@Override
public void start() {
// 创建两个线程组
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// 创建服务端的启动对象
ServerBootstrap serverBootstrap = new ServerBootstrap()
// 设置两个线程组
.group(bossGroup, workerGroup)
// 设置服务端通道实现类型
.channel(NioServerSocketChannel.class)
// 服务端用于接收进来的连接,也就是boosGroup线程, 线程队列大小
.option(ChannelOption.SO_BACKLOG, 100)
.childOption(ChannelOption.SO_KEEPALIVE, true)
// child 通道,worker 线程处理器
.childHandler(new ChannelInitializer<SocketChannel>() {
// 给 pipeline 管道设置自定义的处理器
@Override
public void initChannel(SocketChannel channel) {
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new NettyServerHandler());
}
});
// 绑定端口号,同步启动服务
ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
channel = channelFuture.channel();
// 对关闭通道进行监听,变为同步
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
logger.error("server error.", e);
} finally {
// 释放线程组资源
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
实现具体handler
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
//当通道就绪就会触发该方法
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//进行记录
logger.info("channel active: {}", ctx);
}
//读取数据实际(这里我们可以读取客户端发送的消息)
@Override
public void channelRead(ChannelHandlerContext ctx, MyDataInfo.MyMessage msg) throws Exception {
//将数据读到buffer中
final ByteBuf msgBuf = (ByteBuf) msg;
final byte[] reqBytes = new byte[msgBuf.readableBytes()];
msgBuf.readBytes(reqBytes);
}
//数据读取完毕
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
//使用反射获找到目标方法进行返回
final byte[] respBytes = requestHandler.handleRequest(reqBytes);
ctx.writeAndFlush(respBytes);
}
//处理异常, 一般是需要关闭通道
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
◆ 八.序列化协议
对计算机网络稍微有一点了解的同学都知道,数据在网络中传输是二进制的:01010101010101010,类似这种,只有二进制数据才能在网络中传输。但是在编码之前我们一般先进行序列化,目的是为了优化传输的数据量。因为有的数据太大,需要进行空间优化。
那么我们来区分一下序列化和编码:我画一张图大家都全明白了
定义一个序列化协议,放入作为一个handler放入pipeline中。
Netty支持多种序列化,比如jdk,Json,ProtoBuf 等,这里使用ProtoBuf,其序列化后码流小性能高,非常适合RPC调用。接下来看怎么使用ProtoBuf?
引入Protobuf的依赖
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>2.4.1</version>
</dependency>
序列化:
/**
* 调用对象构造好的Builder,完成属性赋值和序列化操作
* @return
*/
public static byte[] protobufSerializer(){
AnimalProto.Animal.Builder builder = AnimalProto.Animal.newBuilder();
builder.setId(1L);
builder.setName("小猪");
List<String> actions = new ArrayList<>();
actions.add("eat");
actions.add("run");
builder.addAllActions(actions);
return builder.build().toByteArray();
}
反序列化:
/**
* 通过调用parseFrom则完成反序列化
* @param bytes
* @return
* @throws InvalidProtocolBufferException
*/
public static Animal deserialize(byte[] bytes) throws Exception {
AnimalProto.Animal pAnimal = AnimalProto.Animal.parseFrom(bytes);
Animal animal = new Animal();
animal.setId(pAnimal.getId());
animal.setName(pAnimal.getName());
animal.setActions(pAnimal.getActionsList());
return animal;
}
测试:
public static void main(String[] args) throws Exception {
byte[] bytes = serializer();
Animal animal = deserialize(bytes);
System.out.println(animal);
}
以下看到是能正常序列化和反序列化的:
◆ 九.通信协议
通信协议主要是解决网络传输问题,比如TCP拆包粘包问题。
TCP问题:
解决方案:业界的主流协议的解决方案可以归纳如下
这里只是列举出来编码过程,解码是逆过程。(说白了,编码就是找着固定的格式进行写入,解码就是照着固定的格式读)
恭喜你,已经学会写RPC框架了,想深入了解的朋友可以参照源码。进行学习,升级。
来源:https://www.cnblogs.com/monkey-xuan/p/15893604.html
“IT大咖说”欢迎广大技术人员投稿,投稿邮箱:aliang@itdks.com
来都来了,走啥走,留个言呗~
IT大咖说 | 关于版权
由“IT大咖说(ID:itdakashuo)”原创的文章,转载时请注明作者、出处及微信公众号。投稿、约稿、转载请加微信:ITDKS10(备注:投稿),茉莉小姐姐会及时与您联系!
感谢您对IT大咖说的热心支持!