《Dubbo 实现原理与源码解析 —— 精品合集》 《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》 《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》 《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》 《Java 面试题 + Java 学习指南》

摘要: 原创出处 cnblogs.com/2YSP/p/13545217.html 「烟味i」欢迎转载,保留摘要,谢谢!


🙂🙂🙂关注微信公众号:【芋道源码】有福利:

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言将得到认真回复。甚至不知道如何读源码也可以请教噢
  4. 新的源码解析文章实时收到通知。每周更新一篇左右
  5. 认真的源码交流微信群。

一、前言

前段时间看到一篇不错的文章《看了这篇你就会手写RPC框架了》,于是便来了兴趣对着实现了一遍,后面觉得还有很多优化的地方便对其进行了改进。

主要改动点如下:

  1. 除了Java序列化协议,增加了protobuf和kryo序列化协议,配置即用。
  2. 增加多种负载均衡算法(随机、轮询、加权轮询、平滑加权轮询),配置即用。
  3. 客户端增加本地服务列表缓存,提高性能。
  4. 修复高并发情况下,netty导致的内存泄漏问题
  5. 由原来的每个请求建立一次连接,改为建立TCP长连接,并多次复用。
  6. 服务端增加线程池提高消息处理能力

二、介绍

RPC,即 Remote Procedure Call(远程过程调用),调用远程计算机上的服务,就像调用本地服务一样。RPC可以很好的解耦系统,如WebService就是一种基于Http协议的RPC。

图片调用示意图

调用示意图

总的来说,就如下几个步骤:

  1. 客户端(ServerA)执行远程方法时就调用client stub传递类名、方法名和参数等信息。
  2. client stub会将参数等信息序列化为二进制流的形式,然后通过Sockect发送给服务端(ServerB)
  3. 服务端收到数据包后,server stub 需要进行解析反序列化为类名、方法名和参数等信息。
  4. server stub调用对应的本地方法,并把执行结果返回给客户端

所以一个RPC框架有如下角色:

服务消费者

远程方法的调用方,即客户端。一个服务既可以是消费者也可以是提供者。

服务提供者

远程服务的提供方,即服务端。一个服务既可以是消费者也可以是提供者。

注册中心

保存服务提供者的服务地址等信息,一般由zookeeper、redis等实现。

监控运维(可选)

监控接口的响应时间、统计请求数量等,及时发现系统问题并发出告警通知。

三、实现

本RPC框架rpc-spring-boot-starter涉及技术栈如下:

  • 使用zookeeper作为注册中心
  • 使用netty作为通信框架
  • 消息编解码:protostuff、kryo、java
  • spring
  • 使用SPI来根据配置动态选择负载均衡算法等

由于代码过多,这里只讲几处改动点。

3.1 动态负载均衡算法

1.编写LoadBalance的实现类

图片负载均衡算法实现类

2.自定义注解 @LoadBalanceAno

/**
* 负载均衡注解
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface LoadBalanceAno {

String value() default "";
}

/**
* 轮询算法
*/
@LoadBalanceAno(RpcConstant.BALANCE_ROUND)
public class FullRoundBalance implements LoadBalance {

private static Logger logger = LoggerFactory.getLogger(FullRoundBalance.class);

private volatile int index;

@Override
public synchronized Service chooseOne(List<Service> services) {
// 加锁防止多线程情况下,index超出services.size()
if (index == services.size()) {
index = 0;
}
return services.get(index++);
}
}

3.新建在resource目录下META-INF/servers文件夹并创建文件

图片enter description here

4.RpcConfig增加配置项loadBalance

/**
* @author 2YSP
* @date 2020/7/26 15:13
*/
@ConfigurationProperties(prefix = "sp.rpc")
public class RpcConfig {

/**
* 服务注册中心地址
*/
private String registerAddress = "127.0.0.1:2181";

/**
* 服务暴露端口
*/
private Integer serverPort = 9999;
/**
* 服务协议
*/
private String protocol = "java";
/**
* 负载均衡算法
*/
private String loadBalance = "random";
/**
* 权重,默认为1
*/
private Integer weight = 1;

// 省略getter setter
}

5.在自动配置类RpcAutoConfiguration根据配置选择对应的算法实现类

/**
* 使用spi匹配符合配置的负载均衡算法
*
* @param name
* @return
*/
private LoadBalance getLoadBalance(String name) {
ServiceLoader<LoadBalance> loader = ServiceLoader.load(LoadBalance.class);
Iterator<LoadBalance> iterator = loader.iterator();
while (iterator.hasNext()) {
LoadBalance loadBalance = iterator.next();
LoadBalanceAno ano = loadBalance.getClass().getAnnotation(LoadBalanceAno.class);
Assert.notNull(ano, "load balance name can not be empty!");
if (name.equals(ano.value())) {
return loadBalance;
}
}
throw new RpcException("invalid load balance config");
}

@Bean
public ClientProxyFactory proxyFactory(@Autowired RpcConfig rpcConfig) {
ClientProxyFactory clientProxyFactory = new ClientProxyFactory();
// 设置服务发现着
clientProxyFactory.setServerDiscovery(new ZookeeperServerDiscovery(rpcConfig.getRegisterAddress()));

// 设置支持的协议
Map<String, MessageProtocol> supportMessageProtocols = buildSupportMessageProtocols();
clientProxyFactory.setSupportMessageProtocols(supportMessageProtocols);
// 设置负载均衡算法
LoadBalance loadBalance = getLoadBalance(rpcConfig.getLoadBalance());
clientProxyFactory.setLoadBalance(loadBalance);
// 设置网络层实现
clientProxyFactory.setNetClient(new NettyNetClient());

return clientProxyFactory;
}

3.2 本地服务列表缓存

使用Map来缓存数据

/**
* 服务发现本地缓存
*/
public class ServerDiscoveryCache {
/**
* key: serviceName
*/
private static final Map<String, List<Service>> SERVER_MAP = new ConcurrentHashMap<>();
/**
* 客户端注入的远程服务service class
*/
public static final List<String> SERVICE_CLASS_NAMES = new ArrayList<>();

public static void put(String serviceName, List<Service> serviceList) {
SERVER_MAP.put(serviceName, serviceList);
}

/**
* 去除指定的值
* @param serviceName
* @param service
*/
public static void remove(String serviceName, Service service) {
SERVER_MAP.computeIfPresent(serviceName, (key, value) ->
value.stream().filter(o -> !o.toString().equals(service.toString())).collect(Collectors.toList())
);
}

public static void removeAll(String serviceName) {
SERVER_MAP.remove(serviceName);
}


public static boolean isEmpty(String serviceName) {
return SERVER_MAP.get(serviceName) == null || SERVER_MAP.get(serviceName).size() == 0;
}

public static List<Service> get(String serviceName) {
return SERVER_MAP.get(serviceName);
}
}

ClientProxyFactory,先查本地缓存,缓存没有再查询zookeeper。

/**
* 根据服务名获取可用的服务地址列表
* @param serviceName
* @return
*/
private List<Service> getServiceList(String serviceName) {
List<Service> services;
synchronized (serviceName){
if (ServerDiscoveryCache.isEmpty(serviceName)) {
services = serverDiscovery.findServiceList(serviceName);
if (services == null || services.size() == 0) {
throw new RpcException("No provider available!");
}
ServerDiscoveryCache.put(serviceName, services);
} else {
services = ServerDiscoveryCache.get(serviceName);
}
}
return services;
}

问题: 如果服务端因为宕机或网络问题下线了,缓存却还在就会导致客户端请求已经不可用的服务端,增加请求失败率。 解决方案:由于服务端注册的是临时节点,所以如果服务端下线节点会被移除。只要监听zookeeper的子节点,如果新增或删除子节点就直接清空本地缓存即可。

DefaultRpcProcessor

/**
* Rpc处理者,支持服务启动暴露,自动注入Service
* @author 2YSP
* @date 2020/7/26 14:46
*/
public class DefaultRpcProcessor implements ApplicationListener<ContextRefreshedEvent> {



@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
// Spring启动完毕过后会收到一个事件通知
if (Objects.isNull(event.getApplicationContext().getParent())){
ApplicationContext context = event.getApplicationContext();
// 开启服务
startServer(context);
// 注入Service
injectService(context);
}
}

private void injectService(ApplicationContext context) {
String[] names = context.getBeanDefinitionNames();
for(String name : names){
Class<?> clazz = context.getType(name);
if (Objects.isNull(clazz)){
continue;
}

Field[] declaredFields = clazz.getDeclaredFields();
for(Field field : declaredFields){
// 找出标记了InjectService注解的属性
InjectService injectService = field.getAnnotation(InjectService.class);
if (injectService == null){
continue;
}

Class<?> fieldClass = field.getType();
Object object = context.getBean(name);
field.setAccessible(true);
try {
field.set(object,clientProxyFactory.getProxy(fieldClass));
} catch (IllegalAccessException e) {
e.printStackTrace();
}
// 添加本地服务缓存
ServerDiscoveryCache.SERVICE_CLASS_NAMES.add(fieldClass.getName());
}
}
// 注册子节点监听
if (clientProxyFactory.getServerDiscovery() instanceof ZookeeperServerDiscovery){
ZookeeperServerDiscovery serverDiscovery = (ZookeeperServerDiscovery) clientProxyFactory.getServerDiscovery();
ZkClient zkClient = serverDiscovery.getZkClient();
ServerDiscoveryCache.SERVICE_CLASS_NAMES.forEach(name ->{
String servicePath = RpcConstant.ZK_SERVICE_PATH + RpcConstant.PATH_DELIMITER + name + "/service";
zkClient.subscribeChildChanges(servicePath, new ZkChildListenerImpl());
});
logger.info("subscribe service zk node successfully");
}

}

private void startServer(ApplicationContext context) {
...

}
}

ZkChildListenerImpl

/**
* 子节点事件监听处理类
*/
public class ZkChildListenerImpl implements IZkChildListener {

private static Logger logger = LoggerFactory.getLogger(ZkChildListenerImpl.class);

/**
* 监听子节点的删除和新增事件
* @param parentPath /rpc/serviceName/service
* @param childList
* @throws Exception
*/
@Override
public void handleChildChange(String parentPath, List<String> childList) throws Exception {
logger.debug("Child change parentPath:[{}] -- childList:[{}]", parentPath, childList);
// 只要子节点有改动就清空缓存
String[] arr = parentPath.split("/");
ServerDiscoveryCache.removeAll(arr[2]);
}
}

3.3 nettyClient支持TCP长连接

这部分的改动最多,先增加新的sendRequest接口。

图片添加接口

实现类NettyNetClient

/**
* @author 2YSP
* @date 2020/7/25 20:12
*/
public class NettyNetClient implements NetClient {

private static Logger logger = LoggerFactory.getLogger(NettyNetClient.class);

private static ExecutorService threadPool = new ThreadPoolExecutor(4, 10, 200,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), new ThreadFactoryBuilder()
.setNameFormat("rpcClient-%d")
.build());

private EventLoopGroup loopGroup = new NioEventLoopGroup(4);

/**
* 已连接的服务缓存
* key: 服务地址,格式:ip:port
*/
public static Map<String, SendHandlerV2> connectedServerNodes = new ConcurrentHashMap<>();

@Override
public byte[] sendRequest(byte[] data, Service service) throws InterruptedException {
....
return respData;
}

@Override
public RpcResponse sendRequest(RpcRequest rpcRequest, Service service, MessageProtocol messageProtocol) {

String address = service.getAddress();
synchronized (address) {
if (connectedServerNodes.containsKey(address)) {
SendHandlerV2 handler = connectedServerNodes.get(address);
logger.info("使用现有的连接");
return handler.sendRequest(rpcRequest);
}

String[] addrInfo = address.split(":");
final String serverAddress = addrInfo[0];
final String serverPort = addrInfo[1];
final SendHandlerV2 handler = new SendHandlerV2(messageProtocol, address);
threadPool.submit(() -> {
// 配置客户端
Bootstrap b = new Bootstrap();
b.group(loopGroup).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline
.addLast(handler);
}
});
// 启用客户端连接
ChannelFuture channelFuture = b.connect(serverAddress, Integer.parseInt(serverPort));
channelFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
connectedServerNodes.put(address, handler);
}
});
}
);
logger.info("使用新的连接。。。");
return handler.sendRequest(rpcRequest);
}
}
}

每次请求都会调用sendRequest()方法,用线程池异步和服务端创建TCP长连接,连接成功后将SendHandlerV2缓存到ConcurrentHashMap中方便复用,后续请求的请求地址(ip+port)如果在connectedServerNodes中存在则使用connectedServerNodes中的handler处理不再重新建立连接。

SendHandlerV2

/**
* @author 2YSP
* @date 2020/8/19 20:06
*/
public class SendHandlerV2 extends ChannelInboundHandlerAdapter {

private static Logger logger = LoggerFactory.getLogger(SendHandlerV2.class);

/**
* 等待通道建立最大时间
*/
static final int CHANNEL_WAIT_TIME = 4;
/**
* 等待响应最大时间
*/
static final int RESPONSE_WAIT_TIME = 8;

private volatile Channel channel;

private String remoteAddress;

private static Map<String, RpcFuture<RpcResponse>> requestMap = new ConcurrentHashMap<>();

private MessageProtocol messageProtocol;

private CountDownLatch latch = new CountDownLatch(1);

public SendHandlerV2(MessageProtocol messageProtocol,String remoteAddress) {
this.messageProtocol = messageProtocol;
this.remoteAddress = remoteAddress;
}

@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
this.channel = ctx.channel();
latch.countDown();
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
logger.debug("Connect to server successfully:{}", ctx);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
logger.debug("Client reads message:{}", msg);
ByteBuf byteBuf = (ByteBuf) msg;
byte[] resp = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(resp);
// 手动回收
ReferenceCountUtil.release(byteBuf);
RpcResponse response = messageProtocol.unmarshallingResponse(resp);
RpcFuture<RpcResponse> future = requestMap.get(response.getRequestId());
future.setResponse(response);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
logger.error("Exception occurred:{}", cause.getMessage());
ctx.close();
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
logger.error("channel inactive with remoteAddress:[{}]",remoteAddress);
NettyNetClient.connectedServerNodes.remove(remoteAddress);

}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
super.userEventTriggered(ctx, evt);
}

public RpcResponse sendRequest(RpcRequest request) {
RpcResponse response;
RpcFuture<RpcResponse> future = new RpcFuture<>();
requestMap.put(request.getRequestId(), future);
try {
byte[] data = messageProtocol.marshallingRequest(request);
ByteBuf reqBuf = Unpooled.buffer(data.length);
reqBuf.writeBytes(data);
if (latch.await(CHANNEL_WAIT_TIME,TimeUnit.SECONDS)){
channel.writeAndFlush(reqBuf);
// 等待响应
response = future.get(RESPONSE_WAIT_TIME, TimeUnit.SECONDS);
}else {
throw new RpcException("establish channel time out");
}
} catch (Exception e) {
throw new RpcException(e.getMessage());
} finally {
requestMap.remove(request.getRequestId());
}
return response;
}
}

RpcFuture

package cn.sp.rpc.client.net;

import java.util.concurrent.*;

/**
* @author 2YSP
* @date 2020/8/19 22:31
*/
public class RpcFuture<T> implements Future<T> {

private T response;
/**
* 因为请求和响应是一一对应的,所以这里是1
*/
private CountDownLatch countDownLatch = new CountDownLatch(1);
/**
* Future的请求时间,用于计算Future是否超时
*/
private long beginTime = System.currentTimeMillis();

@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
}

@Override
public boolean isCancelled() {
return false;
}

@Override
public boolean isDone() {
if (response != null) {
return true;
}
return false;
}

/**
* 获取响应,直到有结果才返回
* @return
* @throws InterruptedException
* @throws ExecutionException
*/
@Override
public T get() throws InterruptedException, ExecutionException {
countDownLatch.await();
return response;
}

@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
if (countDownLatch.await(timeout,unit)){
return response;
}
return null;
}

public void setResponse(T response) {
this.response = response;
countDownLatch.countDown();
}

public long getBeginTime() {
return beginTime;
}
}

此处逻辑,第一次执行 SendHandlerV2#sendRequest() 时channel需要等待通道建立好之后才能发送请求,所以用CountDownLatch来控制,等待通道建立。 自定义Future+requestMap缓存来实现netty的请求和阻塞等待响应,RpcRequest对象在创建时会生成一个请求的唯一标识requestId,发送请求前先将RpcFuture缓存到requestMap中,key为requestId,读取到服务端的响应信息后(channelRead方法),将响应结果放入对应的RpcFuture中。 SendHandlerV2#channelInactive() 方法中,如果连接的服务端异常断开连接了,则及时清理缓存中对应的serverNode。

四、压力测试

测试环境:

  • (英特尔)Intel(R) Core(TM) i5-6300HQ CPU @ 2.30GHz 4核
  • windows10家庭版(64位)
  • 16G内存

1.本地启动zookeeper 2.本地启动一个消费者,两个服务端,轮询算法 3.使用ab进行压力测试,4个线程发送10000个请求

ab -c 4 -n 10000 http://localhost:8080/test/user?id=1

测试结果

图片测试结果

从图片可以看出,10000个请求只用了11s,比之前的130+秒耗时减少了10倍以上。

五、下载源码

首先点击右下方在看,再长按下方名片关注哦,并后台回复【p017】即可下载。

二维码

文章目录
  1. 1. 一、前言
  2. 2. 二、介绍
    1. 2.0.1. 服务消费者
    2. 2.0.2. 服务提供者
    3. 2.0.3. 注册中心
    4. 2.0.4. 监控运维(可选)
  • 3. 三、实现
    1. 3.1. 3.1 动态负载均衡算法
    2. 3.2. 3.2 本地服务列表缓存
    3. 3.3. 3.3 nettyClient支持TCP长连接
  • 4. 四、压力测试
  • 5. 五、下载源码