必威体育Betway必威体育官网
当前位置:首页 > IT技术

RPC框架pigeon源码分析

时间:2019-07-03 00:43:25来源:IT技术作者:seo实验室小编阅读:70次「手机版」
 

pigeon

pigeon是一个分布式服务通信框架(RPC),是美团点评最基础的底层框架之一。已开源,链接:HTTPs://github.com/dianping/pigeon

从接下来三个方面来分析pigeon的源码。

一. 基础框架

1.1 rpc的基础架构

rpc最基础的架构图

1.2 rpc的基本流程

客户端在调用某一个服务时,这个服务实际上是通过动态代理生成的一个代理类的对象。因此在执行方法的时候,实际上执行的是Invocationhandler的invoke方法(pigeon的InvocationHandler是ServiceInvocationProxy)。然后调用的信息去zk注册中心去拿服务提供方的集群信息,通过负载均衡发现一台实际的服务提供方的服务器地址。将请求信息序列化为二进制数据,然后通过netty的client将请求发送给服务提供方,同时wait服务的响应。

在服务方,启动应用后,rpc将需要发布的服务注册到zk上,开启netty server监听器。服务方收到客户端的数据后,将数据反序列化为请求对象,然后解析请求,进行一系列的过滤操作。最后根据请求的信息定位到服务方唯一的一个服务,执行服务的方法。将执行的结果反序列化为二进制数据,回写到调用服务的客户端。

客户端接收到服务方的响应后,将响应结果反序列化为响应对象,最后返回给用户线程,完成了rpc的调用过程。

1.3 pigeon的主要组件

1.3.1 客户端的组件

ReferenceBean/ProxyBeanFactory:获取服务代理对象的类,实现了FactoryBean,在init方法中初始化了代理对象的创建, 一系列初始化的操作。

InvokerBootStrap:客户端一系列的初始化操作,如初始化ServiceInvocationRepository(令牌桶算法,慢慢放入流量),初始化客户端服务调度器工厂,初始化序列化工厂,初始化负载均衡管理器,初始化路由策略管理器,初始化监控器monitor,初始化response处理器工厂等。

ServiceInvocationProxy:pigeon的动态代理handler,实现了jdk的InvocationHandler接口,每一个客户端动态生成service都会执行ServiceInvocationProxy的invoke方法,在这个方法中service去请求真正的远程服务,这是rpc实现的基础。

ServiceInvocationHandler:服务真正调度的handler,由InvokerProcessHandlerFactory生成,与下面的ServiceInvocationFilter共同组成责任链模式,在handle方法中实际上执行是ServiceInvocationFilter的invoke方法,同时又将下一个handler和上下文Invocationcontext传入filter中,这样实现一层一层的调用。

ServiceInvocationFilter:pigeon的rpc调用的过滤器,客户端有RemoteCallMonitorInvokeFilter,TraceFilter,DegradationFilter,FaultInjectionFilter,clusterInvokeFilter,GatewayInvokeFilter,ContextPrepareInvokeFilter,SecurityFilter,RemoteCallInvokeFilter这些Filter,实现了monitor监控、调用跟踪、服务降级、集群重试、网关、上下文初始化/解析、安全控制、网络调用等过程,实际上rpc的主要功能都是通过这些ServiceInvocationFilter实现的。

NettyClient:pigeon用netty实现的网络客户端,实现了Client接口(还有HttpInvokerClient实现了Client接口,负责http通信),负责初始化netty的ClientBootstrap,维护ChannelPool,以及最重要的给服务提供方write请求数据。

NettyClientHandler:pigeon绑定了ClientBootstrap的客户端ChannelHandler,客户端的众多ChannelHandler之一,绑定在ClientBootstrap的ChannelPipeline上,当服务提供方回复请求结果给客户端时,NettyClientHandler会接收到数据,通过ResponseProcessor将响应数据放到CallbackFuture(盛放结果的类)中,当客户端去取response时,如果已经收到服务提供方响应的数据,则直接获取;如果没有,则await,期间如果收到响应数据,则notify获取响应数据的线程去取,如果超时则报TimeoutException。

1.3.2 服务方的组件

ServiceBean/ServiceRegistry:跟客户端的ReferenceBean类似,ServiceBean是服务方的入口,实现了APPlicationListener了(为了检测服务方的服务是否发布完成),在init方法中完成了provider的初始化和服务注册操作。

ProviderBootStrap:服务端的一系列初始化的操作,如初始化服务方处理器工厂(初始化服务方的handler),初始化序列化工厂,初始化注册管理器,初始化JettyHttpServer(就是我们经常用的那个4080端口的server)等。然后,在ProviderBootStrap的startup方法中初始化NettyServer,执行NettyServer的doStart方法,绑定ServerBootstrap。并初始化requestThreadPoolProcessor的所有线程。

ServicePublisher:服务方注册服务节点信息到zk注册中心的组件,通过RegistryManager管理器注册,最后实际上是使用CuratorRegistry注册器(用CuratorFramework实现的)实现的。

NettyServer:pigeon用netty实现的网络服务端,实现了Server接口(Server还有JettyHttpServer子类,负责http通信,监听了4080端口)。负责初始化服务端的ServerBootstrap,监听tcp端口,绑定ChannelHandler,维护tcp的通信。

NettyServerHandler:pigeon绑定了ServerBootstrap的服务端ChannelHandler,服务端的众多ChannelHandler之一,当客户端发起数据请求时,NettyServerHandler会接收到数据,如果是心跳检查,直接处理回写数据,如果是真实的请求,则通过RequestThreadPoolProcessor去处理,找到服务调用然后回写数据。

RequestThreadPoolProcessor:根据不同的messageType选择不同的ServiceInvocationHandler,提供了四种类型的handler,分别是业务处理handler、心跳处理handler、健康检查handler、scanner心跳handler。内部维护了一组线程池,提供了服务隔离功能。通过ServiceInvocationHandler和ServiceInvocationFilter去处理请求信息,在BusinessProcessFilter通过反射执行真正的服务方法,在WriteResponseProcessFilter回写处理完成的response数据给客户端,完成整个服务端处理请求的链路。

二. 客户端如何调用pigeon的服务

客户端调用pigeon服务主要分为两步:

第一步:初始化及启动invoker,获取动态代理类对象;

第二步:通过代理类对象动态的调用远程服务,并将响应结果返回给客户端。

2.1 invoker初始化的处理链路

客户端创建代理服务对象有两种方式:Spring注入和api编码的方式。如果是spring依赖注入的方式,是通过ProxyBeanFactory或ProxyBeanFactory获取动态代理的bean,实际上是通过ServiceFactory的getService()方法调用的;如果是api编程的方式,则是直接通过ServiceFactory的getService()获取的。

在ServiceFactory中,调用了AbstractServiceProxy的getProxy方法,invoker初始化大部分工作都是在这个类中完成的。首先调用invoker初始化器InvokerBootStrap初始化一系列的管理器与工厂(调度仓库、调度器、序列化工厂、负载均衡管理器、路由策略管理器、监控器、response处理器等等),这是客户端调用rpc服务必须的组件。

然后会调用Abstractserializer的proxyRequest方法通过Proxy.newProxyinstance创建一个代理类对象,设置InvocationHandler为ServiceInvocationProxy,返回创建好的proxy。

接下来进行一系列的注册,先后注册路由策略和服务配置,在ClientManager中根据invokerConfig去zk注册中心上拉取远程服务器的ip和端口号。在CuratorRegistry中通过CuratorClient(就是CuratorFramework实现的)从zk上拉取address,同时构建NettyClient也是在这里实现的。

最后将proxy服务 put到AbstractServiceProxy的services(ConcurrentHashMap类型)中,然后返回代理对象,完成整个过程。

2.2 调用远程服务的处理链路

当客户端通过proxyService调用远程服务时,由于proxyService是代理类,因此实际上会去调用ServiceInvocationProxy的invoke方法,在invoke方法中,会执行ServiceInvocationHandler的handle方法。

ServiceInvocationHandler与ServiceInvocationFilter共同组成责任链模式,ServiceInvocationHandler的handle方法实际上执行的是ServiceInvocationFilter的invoke方法。InvokerProcessHandlerFactory构建了一个filterList,遍历filterList,对每一个filter都创建了一个ServiceInvocationHandler,同时将上一个创建的ServiceInvocationHandler传入filter的invoke方法中。返回最后创建的handler。这样如果在filter中需要调用下一个filter,执行传入的handler.handle()即可。

filterList的代码如下:

public static void init() {

   if (!isInitialized) {

      if (constants.MONITOR_ENABLE) {

         registerBizProcessFilter(new RemoteCallMonitorInvokeFilter());

      }

      registerBizProcessFilter(new TraceFilter());

      registerBizProcessFilter(new DegradationFilter());

      registerBizProcessFilter(new FaultInjectionFilter());

      registerBizProcessFilter(new ClusterInvokeFilter());

      registerBizProcessFilter(new GatewayInvokeFilter());

      registerBizProcessFilter(new ContextPrepareInvokeFilter());

      registerBizProcessFilter(new SecurityFilter());

      registerBizProcessFilter(new RemoteCallInvokeFilter());

    //最后返回的是执行RemoteCallMonitorInvokeFilter的handler,执行链到RemoteCallInvokeFilter为止

      bizInvocationHandler = createInvocationHandler(bizProcessFilters);

      isInitialized = true;

   }

}

最后执行的ServiceInvocationFilter是RemoteCallInvokeFilter,实现发送请求的地方,通过NettyClient将请求发送到远程服务的机器上去。RemoteCallInvokeFilter提供了四种调用方式:SYNC,CALLBACK,FUTURE,ONEWAY。这就是我们在spring配置中经常需要设置的参数

四种调用方式都是通过InvokerUtils的sendRequest方法中调用NettyClient的write方法,将请求数据发送到目的服务器去。

  • SYNC:同步的调用方式,发送完数据后,通过getResponse去取结果,如果远程服务已经返回结果了就直接取,还没有返回结果就等待,等待超时了就报TimeoutException异常。
  • CALLBACK:回调的调用方式,把回调函数传入到方法中,当远程服务返回了结果会调用回调函数,异步的调用。
  • FUTURE:future的调用方式,发送完请求后,创建一个future对象放入当前线程的threadlocal中,客户端需要的话可以通过threadLocal去取结果。对pigeon来说是异步的调用,如果客户端需要取结果则是非阻塞同步的方式。
  • ONEWAY:单向调用的方式,发送完请求直接返回,不关心处理结果,异步的调用。

调用方法如下:

switch (callMethod) {

    case SYNC:

        CallbackFuture future = new CallbackFuture();

        //在InvokerUtils的sendRequest方法中调用NettyClient的write方法,将请求数据发送到目的服务器去

        response = InvokerUtils.sendRequest(client, invocationContext.getRequest(), future);

        invocationContext.getTimeline().add(new TimePoint(TimePhase.Q));

        if (response == null) {

            //如果是同步的,就通过getResponse去取结果,远程服务还没有返回结果就等待,有就直接取

            response = future.getResponse(request.getTimeout());

        }

        break;

    case CALLBACK:

        InvocationCallback callback = invokerConfig.getCallback();

        InvocationCallback tlCallback = InvokerHelper.getCallback();

        if (tlCallback != null) {

            callback = tlCallback;

            InvokerHelper.clearCallback();

        }

        //回调的方式则将你的回调函数传入,远程服务返回结果后会调用回调函数

        InvokerUtils.sendRequest(client, invocationContext.getRequest(), new ServiceCallbackwrapper(

                invocationContext, callback));

        response = NO_RETURN_RESPONSE;

        invocationContext.getTimeline().add(new TimePoint(TimePhase.Q));

        break;

    case FUTURE:

        ServiceFutureImpl futureImpl = new ServiceFutureImpl(invocationContext, request.getTimeout());

        InvokerUtils.sendRequest(client, invocationContext.getRequest(), futureImpl);

        //future方式,将future对象放入当前线程的threadLocal中

        FutureFactory.setFuture(futureImpl);

        //返回的是一个future的response,并没有真是的结果

        response = InvokerUtils.createFutureResponse(futureImpl);

        invocationContext.getTimeline().add(new TimePoint(TimePhase.Q));

        break;

    case ONEWAY:

        //oneway不处理返回结果

        InvokerUtils.sendRequest(client, invocationContext.getRequest(), null);

        response = NO_RETURN_RESPONSE;

        invocationContext.getTimeline().add(new TimePoint(TimePhase.Q));

        break;

    default:

        throw new badrequestException("Call type[" + callMethod.getName() + "] is not supported!");

}

将请求数据发送给远程服务的机器后,客户端需要接受服务器响应的结果。这个是通过NettyClientHandler等ChannelHandler实现。在创建NettyClient的时候,ClientBootstrap会setPipelineFactory将一组ChannelHandler关联到Client上去。当服务端给客户端发送数据时,会通过这一组handler来响应处理。handlers解决了tcp粘包、序列化、反序列化等问题。

代码如下:

public ChannelPipeline getPipeline() throws Exception {

    ChannelPipeline pipeline = pipeline();

    pipeline.addLast("framePrepender"new FramePrepender());

    pipeline.addLast("frameDecoder"new FrameDecoder());

    pipeline.addLast("crc32Handler"new Crc32Handler(codecConfig));

    pipeline.addLast("compressHandler"new CompressHandler(codecConfig));

    pipeline.addLast("invokerDecoder"new InvokerDecoder());

    pipeline.addLast("invokerEncoder"new InvokerEncoder());

    pipeline.addLast("clientHandler"new NettyClientHandler(this.client));

    return pipeline;

}

当收到远程服务的响应数据时,NettyClientHandler会调用ResponseThreadPoolProcessor使用多线程去处理response,在线程中,使用ServiceInvocationRepository的单例对象处理response,把response放到RemoteCallInvokeFilter创建的CallbackFuture中,用condition.signal()唤醒RemoteCallInvokeFilter等待的线程(如果是同步的调用),完成整个调用工作。

pigeon的调用结构图

三. 服务端如何处理rpc的请求

服务提供方提供和处理pigeon服务主要分为两步:

第一步:初始化及启动provider,监听tcp端口,注册服务;

第二步:监听客户端发送的请求消息,处理结果并返回给客户端。

3.1 provider初始化的处理链路

服务端初始化provider和注册服务也有两种方式:spring注入和api编码的方式。如果是spring注入的方式,通过ServiceBean的init方法,将coder设置在services中的服务通过ServiceFactory.addServices()方法初始化和注册;如果是api编码的方式,则是直接调用ServiceFactory.addService()的方法进行初始化和注入,过程都是相同的。

在ServiceFactory中,ServiceFactory的静态代码块会执行ProviderBootStrap.init()方法,ProviderBootStrap是服务方的启动器,完成服务方各种初始化工厂和管理器(服务方处理器工厂、序列化工厂、注册管理器),还会启动JettyHttpServer监听4080端口。跟客户端的InvokerBootStrap类似,是服务方提供rpc服务必须的组件。

接下来调用PublishPolicy的doAddService()方法,PublishPolicy首先去startup ProviderBootStrap,在这里才初始化了NettyServer,执行NettyServer的doStart方法,绑定ServerBootstrap,监听tcp端口(并不是固定端口号)。并初始化RequestThreadPoolProcessor的所有线程。然后通过ServicePublisher发布注册服务。

ServicePublisher通过publishService发布服务,实际上是调用preparePublishTask()方法,通过自身的静态内部类PublishTask在线程中异步发布注册服务,在PublishTask通过RegistryManager的registerService()方法在CuratorRegistry注册了服务,而在CuratorRegistry中通过CuratorClient(CuratorFramework实现的)实现了zk注册服务,基本逻辑是:如果当前serviceName在zk注册中心的节点上存在,则在这个节点添加一个address,如果不存在,则创建一个节点存放address。

3.2 响应请求消息的处理链路

上面的启动provider的过程中,NettyServer将一组ChannelHandler通过setPipelineFactory关联到ServerBootstrap上去。客户端发起远程服务请求后,会通过这一组handler来响应处理。与客户端的handlers类似,服务端的handlers同样解决了tcp粘包、序列化、反序列化等问题。

public ChannelPipeline getPipeline() {

    ChannelPipeline pipeline = pipeline();

    pipeline.addLast("framePrepender"new FramePrepender());

    pipeline.addLast("frameDecoder"new FrameDecoder());

    pipeline.addLast("crc32Handler"new Crc32Handler(codecConfig));

    pipeline.addLast("compressHandler"new CompressHandler(codecConfig));

    pipeline.addLast("providerDecoder"new ProviderDecoder());

    pipeline.addLast("providerEncoder"new ProviderEncoder());

    pipeline.addLast("serverHandler"new NettyServerHandler(server));

    return pipeline;

}

NettyServerHandler收到客户端的请求数据后,首先判断是不是scanner心跳的请求, 如果是,则直接回写数据回去即可;如果是客户端的正常请求,则通过AbstractServer的processRequest()方法去处理。

AbstractServer直接通过RequestThreadPoolProcessor的doProcessRequest来处理请求数据。RequestThreadPoolProcessor内部维护了一组线程池,提供了服务隔离功能。RequestThreadPoolProcessor首先根据请求去获取对应的线程池(默认提供slowRequestProcessThreadPool、sharedRequestProcessThreadPool,另外还提供了可定制针对单个服务和方法的methodThreadPools和serviceThreadPools),然后检测请求是否可行。如果一切ok,则创建一个Callable对象,用线程池去处理这个请求。

在Callable中,首先通过ProviderProcessHandlerFactory的selectInvocationHandler()方法创建一个handler,与客户端相同的是,ServiceInvocationHandler与ServiceInvocationFilter同样构成责任链模式,provider提供四种handler链,分别是业务处理handler、心跳处理handler、健康检查handler、scanner心跳handler,根据不同的请求处理不同的handler链。

责任链模式使得ServiceInvocationFilter一层一层的调用,直到BusinessProcessFilter为止开始一层一层的返回。BusinessProcessFilter是FilterList最核心的ServiceInvocationFilter,在这个Filter中,provider通过请求的数据获取到真正对应的bizService和method,然后通过反射的方式调用这个方法,返回调用结果,把结果封装到response中。

InvocationResponse response = null;

ServiceMethod method = invocationContext.getServiceMethod();

if (method == null) {

    method = ServiceMethodFactory.getMethod(request);

}

ProviderHelper.setContext(invocationContext);

invocationContext.getTimeline().add(new TimePoint(TimePhase.M, system.currenttimemillis()));

Object returnObj = null;

try {

    returnObj = method.invoke(request.getparameters());

finally {

    ProviderHelper.clearContext();

    if (Constants.REPLY_MANUAL || invocationContext.isAsync()) {

        if (request.getCallType() == Constants.CALLTYPE_REPLY) {

            request.setCallType(Constants.CALLTYPE_MANUAL);

        }

    }

}

invocationContext.getTimeline().add(new TimePoint(TimePhase.M, System.currentTimeMillis()));

if (request.getCallType() == Constants.CALLTYPE_REPLY) {

    response = ProviderUtils.createSuccessResponse(request, returnObj);

}

return response;

FilterList中还有一个WriteResponseProcessFilter,WriteResponseProcessFilter会通过netty的channel将处理结果response发送给客户端,这样就完成了服务端处理请求的过程。

 

四. pigeon中的设计模式

pigeon是一个设计很优雅的rpc框架,里面用到很多很好的设计模式

4.1 代理模式

客户端创建代理服务对象的时候就用到了代理模式,创建的对象并不是通过new的方式来创建,而是Proxy.newProxyInstance()方法。在调用代理类的方法时,会去调用传入newProxyInstance的参数ServiceInvocationProxy对象的invoke方法,这样就实现了通过网络通信去调用远程服务的功能了。

pigeon在AbstractSerializer类中通过代理模式创建了代理服务对象。

4.2 工厂模式

工厂模式是将创建对象的操作放在工程类中,需要使用的时候去工厂类中取。pigeon使用的工厂模式很多,如ProviderProcessHandlerFactory,在这个工厂类中创建了四种ServiceInvocationHandler,需要用的时候根据不同的type去取。如下:

public static ServiceInvocationHandler selectInvocationHandler(int messageType) {

   if (Constants.MESSAGE_TYPE_HEART == messageType) {

      return heartbeatInvocationHandler;

   else if (Constants.MESSAGE_TYPE_HEALTHCHECK == messageType) {

      return healthCheckInvocationHandler;

   else if (Constants.MESSAGE_TYPE_SCANNER_HEART == messageType) {

      return scannerHeartBeatInvocationHandler;

   else {

      return bizInvocationHandler;

   }

}

4.3 策略模式

策略模式是定义了一系列的策略,并将每一个策略封装起来,而且使他们可以相互替换,在运行时动态选择具体要执行的行为。pigeon也使用了很多策略模式。如在ClusterInvokeFilter中,定义了一组集群重试策略,在运行时可以动态的选择具体需要执行的集群重试行为。

public InvocationResponse invoke(ServiceInvocationHandler handler, InvokerContext invocationContext)

      throws throwable {

   InvokerConfig<?> invokerConfig = invocationContext.getInvokerConfig();

   Cluster cluster = ClusterFactory.selectCluster(invokerConfig.getCluster());

   if (cluster == null) {

      throw new illegalargumentException("Unsupported cluster type:" + cluster);

   }

   return cluster.invoke(handler, invocationContext);

}

4.4 责任链模式

责任链模式是使多个对象都有机会处理请求,从而避免请求的发送者和接受者之间的耦合关系, 将这个对象连成一条链,并沿着这条链传递该请求,直到有一个对象处理他为止。在pigeon中,ServiceInvocationHandler与ServiceInvocationFilter组成责任链模式,InvokerProcessHandlerFactory构建了一个filterList,ServiceInvocationHandler的handle方法实际上执行的是ServiceInvocationFilter的invoke方法。ServiceInvocationFilter一层一层的调用,直到最后一个filter然后开始一层一层的返回。

for (int i = filterList.size() - 1; i >= 0; i--) {

   final V filter = filterList.get(i);

   final ServiceInvocationHandler next = last;

   last = new ServiceInvocationHandler() {

      @SuppressWarnings("unchecked")

      @Override

      public InvocationResponse handle(InvocationContext invocationContext) throws Throwable {

         InvocationResponse resp = filter.invoke(next, invocationContext);

         return resp;

      }

   };

}

4.5 单例模式

单例模式是最简单的设计模式,也是pigeon中用得最多的设计模式。单例模式可以保证系统中,应用该模式的类一个类只有一个实例。pigeon的资源管理器大部分都设计为单例模式,如ServiceInvocationRepository、RoutePolicyManager、ServiceConfigManager、ClientManager等等。主要是为了管理公共资源,保持唯一访问路径

相关阅读

传统行业互联网化的商业模式分析框架

在互联网的帮助下,传统行业可以再上新台阶。互联网+是中国在2015年最炙手可热的概念。之所以这个概念如此火爆,原因还是在于中国经

cat 分布式框架

大众点评开源分布式监控平台 CAT 深度剖析 一、CAT介绍CAT系统原型和理念来源于eBay的CAL的系统,CAT系统第一代设计者吴其敏在eBay

ssm框架通用代码生成工具

做了好几年的.net,上半年开始做java,发现java框架确实非常牛逼。基本上可以封装成框架的都有成品。相比之下.net确实差很多。但是也

【英雄之旅】故事框架:与其将自己打造成英雄,不如让用户

只有掌握了人类最根本的情感模式,就能说出穿透时空、种族、语言的故事——这也是为什么好的文学和影视作品能卖座全球,因为这些热卖

最全互联网金融6大板块风控研究框架:监管政策、行业模

互联网金融风险主要包括政策法律风险、监管风险、交易风险、技术风险、认知风险等类型,除了对互联网金融企业或客户本身产生直接影

分享到:

栏目导航

推荐阅读

热门阅读