RPC 基础功能

• 数据传输
• 序列化/反序列化
• 客户端代理类
• 请求映射分发

RPC 产品功能

除了RPC基础功能为，RPC还具备八大功能模块
Consumer：连接管理、负载均衡、请求路由、超时处理
Provider：队列/线程池、超时丢弃、优雅关闭、过载保护

Consumer 核心设计

完整的consumer功能模块图

• 请求路由 通过一系列规则过滤出可以选择的provider节点列表，在应用隔离，读写分离灰度发布中发挥关键作用。 匹配规则 规则描述 待比较属性 运算符 属性匹配值 匹配节点 数据结构设计 链表
  以IP分流规则举例
  行为 链表
• 连接管理
  保持与Provider的长连接，用于传输数据和返回结果，通常情况下基于tcp连接，udp&http也可以实现。 初始化时机 饿汉模式-服务间连接、数据库连接 懒汉模式-网关连接 连接数维护 服务连接池 数据库连接池 心跳/断线重连机制 client线程模型
• 负载均衡
  确保多个Provider实例节点流量均匀合理(并非绝对平均)；支持节点扩容缩容与灰度发布(调整流量权重比例) 轮询 —— 按顺序轮流分发请求 随机 —— 随机选择分发请求 取模 —— 请求数 % 实例数得到集群下标，将请求分发给指定下标的集群实例。 权重 —— 根据实例的性能设置权重值进行请求分发 一致性hash —— 根据hash值将 0 - 2^32次方 按照实例数划分为不同的hash范围，根据范围值分发请求
• 超时处理 对于长时间没有响应的请求，需要作异常处理，及时释放资源。 工作线程阻塞位置 等待响应通知 超时逻辑 工作线程等待通知 数据返回终止等待 超时抛出异常 数据结构 Map

Provider 核心设计

队列/线程池

将不同类型的请求，放入各自的队列，每个队列分配独立的线程池，进行资源隔离. 这里的Thread Pool并非指java中的线程池对象，这里是真正的线程池。

Provider 作为服务提供方，对调用方的信息是无感知的，随业务发展可能会有N个consumer对provider进行调用，工作线程有限，因而需要设计数据结构(队列)对请求进行存放。
数据结构设计分析：

• 请求量多大？是否需要削峰填谷缓存请求？
• 是否存在请求顺序性要求？
• 请求之间相互隔离？不同的请求放不同的队列，不同队列使用不同线程池，资源隔离
  线程与线程之间什么样的通信方式是最安全的？
  通过通信方式来共享内存，不要通过共享内存来通信。线程与线程之间通过队列方式通信是最好的。 补充知识： OS分配资源的基本单位是进程，cpu调度的基本单位是线程。

线程分配的合理数：
线程数 = CPU核数 * 2 + 2；线程过多会造成线程等待和上下文切换的资源开销；线程太少又无法充分利用cpu资源。

这里就不提供压测截图了，关于压测有兴趣可以百度自行了解，或者找测试同学沟通
QPS 队列线程 | 2w+ | 5w+ | 8w+ | 11w+

• 单队列多线程 1*64 | 0.00s | 0.01/0.00s | 0.05/0.00s | 0.19/0.01s
• 多队列单线程 64*1 | 0.00s | 0.01/0.00s | 0.02/0.00 | 0.07/0.01s

在QPS 8w+ 时 单队列多线程模型的耗时明显比多队列单线程高

超时丢弃

快速失败已超时的请求，缓解队列的压力

1. IO线程处理反序列化后将请求放入队列（req Queue）

入队代码

publick void run(Group group, IAsyncHandler handler){
    //构造 AsyncTask 对象放入队列
    AsyncTask task = new AsyncTask(taskTimeout,handler);
    balance(group,task);
}

//异步任务构造方法
public AsyncTask(int timeout,IAsyncHandler handler){
    super();
    if(timeout < 0){
      timeout = 1000;
    }
    this.timeout = timeout;//超时时间
    this.handler = handler;
    this.addTime = System.currentTimeMillis();//入队时间
}

2. 当线程池中工作线程从队列中取出请求进行处理时，判断当前请求是否超时。超时则丢弃，未超时则调用服务处理

出队代码

public void run() {
    while(!isStop){
        AsyncTask task  = null;
        try {
            task = taskQueue.poll(1500,TimeUnit.MILLISECONDS);//取出队列任务
            if(null != task){
                execTimeoutTask(task);//判断任务是否超时
            }
        }catch (InterruptedException e){
            logger.error("has error!",e);
        }catch (Throwable ex){
            if(task !=null ){
                task.getHandler().exceptionCaught(ex);
            }
        }
    }
}

public void execTimeoutTask(AsyncTask task) throw {
    //当前时间减去入队时间是否大于超时时间 超时回调超时
    if(System.currentTimeMillis() - task.getAddTime() > task.getTimeout()){
        task.getHandler().exceptionCaught(new TimeoutException(threadFactoryName + "async task timeout!"));
        return;
    } else {
       //没有超时回调处理任务
      Object obj = task.getHandler().run();
      task.getHandler().messageReceived();
    }
}

对于consumer 而言，一个请求调用的超时时间 = 请求队列排队时间（主要耗时） + 程序处理时间

优雅关闭

进程结束前应确保队列中的的请求全部处理完成，这路的处理完成是直接回复 并非处理逻辑。
如何通知调用方？

• 返回数据中带关闭信息 建议
• 专门关闭协议通知调用方 不建议 服务提供方对调用方是无感知的；即便是在知道调用方的情况下这样的设计也会造成循环依赖，且调用可能失败，系统复杂度过高。
  返回数据中带关闭信息可以复用现有RPC通道

优雅关闭Server端实现

• 监听关闭信号 kill -12

@Component
 public class SignalConfig {
     @Autowired
     private SignalRegistry signalRegistry
     public void init() { signalRegistry.register(); }
 }
 //监听关闭信号
 public void register() {
     try {
         if(StringUtils.isNotBlank(osName) && (!isMac() && !isWindows())) {
             Signal sig = new Signal("USR2");//这里的USR2 其实就是 -12 代表的是用户自定义信号
             //这表示当用户通过 kill -12 来 结束某个进程时 回调  operateSignalHandler 函数进行处理
             Signal.handle(sig, operateSignalHandler);
             
             Signal sig2 = new Signal("STKFLT");
             Signal.handle(sig2, breakHeartbeatSignal);
         }
     } catch ( Exception e ) {
         logger.error("------------ signal register failed ! --------------",e)
     }
 }

• 改变服务状态

 //改变服务状态
 @Component
 public class OperateSignal implements SignalHandler {
     private static Logger logger = LoggerFactory.getLogger(OperateSignal.class);
     @Autowired
     private RpcContext rpcContetx;
     @Override
     public void handle(Singal singalName){
         logger.info("server : {} current state is : {}",new Object[] {rpcContext.getServiceName(), rpcContext.getServerState()});
         //设置当前服务状态为重启
         rpcContext.setServerState(ServerStateType.Reboot);
         logger.info("server : {} will reboot !", new Object[]{rpcContext.getServiceName()});
     }
 }

• 通知客户端

//超时处理判断请求没有超时时调用 service方法处理请求
public Object innerInvoke(RpcContext rpcContext, MethodSignature methodSignature) throw Exception {
    //这里使用责任链模式维护了RPC上下文对象
    requestFilter(context);
    Object response = null;
    RPCContext.setThreadLocal(context);
    ..... 省略.....
    context.getResponseProtocol().setSdpEntity(response);
    responseFilter(context);
    RpcContext.clear();
    return context;
}

protected void requestFilter(RpcContext context) throws Excpetion {
 logger.debug("begin requestFilters");
 for(IFilter filter : requestFilters){
     if(context.getExecFilter() == ExecFilterType.ALL ||
        context.getExecFilter() == ExecFilterType.RequestOnly){
        filter.filter(context);
     }
     logger.debug("end requestFilters");
 }
}

public void filter(RpcContext context) throws Exception {
    if(serviceContext.getServerState) == ServerStateType.Reboot &&
    (protocol.getPlatformType() == PlatformType.Java||protocol.getPlatformType() == PlatformType.PHP)) {
        //封装RPC响应
        RpcResponse response  = new RpcResponse();
        ResetProtocol rp = new ResetProtocol();
        rp.setMsg("This server is reboot!");
        responseProtocol.setSdpEntity(rp);
        response.setResponseBuffer(responseProtocol.toBytes(false,null));
        context.setRpcResponse(response);
        context.setExecFilter(ExecFilterType.None);
        //不再继续过滤
        context.setDoInvoke(false);
    }
}

程序启动后初始化，初始化时 初始化容器、初始化插件、初始化监听信号（重要）、初始化服务
解释： 通常结束进程我们用 kill -9 命令 这里的 9 其实就是一种信号量，用户可以自定义信号量发送给操作系统，可以自定义实现

优雅关闭client端实现

• 根据返回改变节点状态

public Protocol request(Protocol requestProtocol) throws Exception {
    //判断服务节点状态
    if(ServerState.Reboot == state || ServerState.Dead == state) {
        throw new RebootException();
    };
    ....
    //注册窗口事件
    socket.registerRec(requestProtocol.getSessionID());
    //异步发送请求
    socket.send(data);
    ....
    
    //接收数据，等待数据到达事件
    byte[] buffer = socket.receive(requestProtocol.getSessionID().currUserCount);
    Protocol receiveProtocol = Protocol.fromBytes(buffer,socket.isRights(),socket.getDESKey());
    
    //接收回包首先判断类型是否是重启
    SDPType sdpType = receiveProtocol.getSDPType();
    if(sdpType == SDPType.Reset){
        //如果是重启 服务节点置为不可用
        this.asReboot();
        logger.info("server: [{}], address : [{}] was rebooted, will choose another one !",
        new Object[]{this.getName(),this.getAddress()});
        //抛出服务重启异常
        throw new RebootException();
    }
}

设置服务器为重启状态

public void asReboot() {
    if(ServerState.Reboot != this.getState()) {
        this.setState(ServerState.Reboot);
        this.setDeadTime(System.currentTimeMillis());
        this.setWeight(-1);
        this.getSocketPool().destroy();
        ServerStateDetector.instance().add(this);
        logger.debug("this server is reboot! host: "+ this.getAddress());
    }
}

• 节点探索

过载保护

服务提供方为保证正常运行，主动丢弃超出处理能力的请求。
这里的主动丢弃与超时丢弃不同。每个请求入队时判断队列中的请求数是否达到阈值，超过则直接丢弃。

入队任务代码

publick void run(Group group, IAsyncHandler handler){
    //构造 AsyncTask 对象放入队列
    AsyncTask task = new AsyncTask(taskTimeout,handler);
    balance(group,task);
}

public void balance(Group group,AsyncTask task) {
    if(group == Group.HIGH) {
        balanceTask(highFactor.getAndIncrement(), groupHighWorkers, task);
    }else if(group == Group.DEFAULT){
        balanceTask(defaultFactor.getAndIncrement(), defaultWorkers, task);
    }
}

public void balanceTask(int factor, AsyncWorker[] workers, AsyncTask task) {
    int idx = factor % workers.length;
    //如果预设值大于0
    if(limitSize>0){
        workers[idx].addTask(task,limitSize,mode);
    }else{
        worker[idx].addTask(task);
    }
}

void addTask(AsyncTask task,int limitSize,boolean abortNewTask) {
    //如果请求队列长度超出了预设值  抛弃新来的任务
    if(this.taskQueue.size() >= limitSize){
        if(abortNewTask) {
            task.getHandler().exceptionCaught(new TimeoutException(threadFactoryName + 
            "abort this task, because the queue is full!"));
        }else{
            elimintateOldTask(task);
        }
    }else{
        this.queue.offer(task);
    }
}