zoukankan      html  css  js  c++  java
  • Dubbo——服务调用过程

    引言

    经过之前文章的铺垫,现在可以来分析服务的交互调用过程了。

    服务的交互

    服务降级

    从名字上看我们不难理解MockClusterInvoker相当于是对普通Invoker(如DubboInvoker)的装饰增强,提供集群容错相关的功能,因此最终还是会进入到DubboInvoker,所以这里就以MockClusterInvoker.invoke方法开始分析:

    public Result invoke(Invocation invocation) throws RpcException {
    	  Result result = null;
           // 获取mock配置
          String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim(); 
          if (value.length() == 0 || value.equalsIgnoreCase("false")){
          	// 未配置mock
          	result = this.invoker.invoke(invocation);
          } else if (value.startsWith("force")) {
          	// 强制执行mock逻辑,不发起远程调用
          	result = doMockInvoke(invocation, null);
          } else {
          	// 首先调用远程服务,失败后执行mock逻辑,不抛出异常
          	try {
          		result = this.invoker.invoke(invocation);
          	}catch (RpcException e) {
    		if (e.isBiz()) {
    			throw e;
    		} else {
    			result = doMockInvoke(invocation, e);
    		}
    	}
          }
          return result;
    }
    

    Mock服务就是提供一个替代的服务接口,也就是服务降级,从这里看出有三种配置方式:直接调用服务直接调用mock服务以及远程服务调用失败后调用mock服务。服务降级这里暂不分析,咱们接着看远程服务调用this.invoker.invoke,这个invoker对象上文分析过了,是FailoverClusterInvoker对象(即调用失败后重试其它远程服务),不过该类中没有invoke方法,所以是调用父类的AbstractClusterInvoker.invoke:

    public Result invoke(final Invocation invocation) throws RpcException {
    
        checkWheatherDestoried();
    
        LoadBalance loadbalance;
        
        List<Invoker<T>> invokers = list(invocation);
        // 获取负载均衡配置,默认使用的是随机算法
        if (invokers != null && invokers.size() > 0) {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
        } else {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        // 具体的容错机制有子类实现
        return doInvoke(invocation, invokers, loadbalance);
    }
    

    这里首先会获取配置的负载均衡算法,默认是使用RandomLoadBalance,不过这个并不是简单的随机,而是加入了权重值,感兴趣的可以去看看其doSelect方法的实现。

    集群容错

    拿到负载均衡器后,会传入到集群容错的的具体实现类中进行选择。Dubbo默认使用的是FailoverClusterInvoker失败切换容错机制,所以这里以此为例,其它的容错机制读者可自行分析。

    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    	List<Invoker<T>> copyinvokers = invokers;
    	// 检查invoker列表是否变更
    	checkInvokers(copyinvokers, invocation);
    	// 获取需要重试的次数,即invoker个数
        int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        // retry loop.
        RpcException le = null; // last exception.
        // 已经调用过的
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
        Set<String> providers = new HashSet<String>(len);
        for (int i = 0; i < len; i++) {
        	//重试时,进行重新选择,避免重试时invoker列表已发生变化.
        	//注意:如果列表发生了变化,那么invoked判断会失效,因为invoker示例已经改变
        	if (i > 0) {
        		checkWheatherDestoried();
        		copyinvokers = list(invocation);
        		//重新检查一下
        		checkInvokers(copyinvokers, invocation);
        	}
        	// 负载均衡筛选出的invoker
            Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers((List)invoked);
            try {
            	// 调用对应的invoker,执行成功返回对应的结果,否则出现异常则重试其它服务
                Result result = invoker.invoke(invocation);
                return result;
            } catch (RpcException e) {
                if (e.isBiz()) { // biz exception.
                    throw e;
                }
                le = e;
            } catch (Throwable e) {
                le = new RpcException(e.getMessage(), e);
            } finally {
                providers.add(invoker.getUrl().getAddress());
            }
        }
        // 重试失败则抛出异常
        throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
                + invocation.getMethodName() + " in the service " + getInterface().getName() 
                + ". Tried " + len + " times of the providers " + providers 
                + " (" + providers.size() + "/" + copyinvokers.size() 
                + ") from the registry " + directory.getUrl().getAddress()
                + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
                + Version.getVersion() + ". Last error is: "
                + (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
    }
    

    该容错机制的逻辑就是根据重试次数进行循环调用,每次循环都会通过select挑选出一个代理对象进行远程调用,若成功则直接返回结果,否则直到达到重试次数抛出异常。select方法中还有一个细节需要注意,我们来看一下:

    protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.size() == 0)
            return null;
        String methodName = invocation == null ? "" : invocation.getMethodName();
        
        // 获取sticky粘滞连接配置,该配置表示是否始终使用同一个服务,除非该服务挂掉
        boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName,Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY) ;
        {
            // invoker列表中已经没有该粘滞连接对象,表示该服务已挂掉
            if ( stickyInvoker != null && !invokers.contains(stickyInvoker) ){
                stickyInvoker = null;
            }
            // 使用粘滞连接需要保证该连接不在invoked列表中,该列表中的服务表示存活且已经调用过的服务,
            // 但由于某种原因未提供正常的服务,粘滞连接也不再去调用该服务
            if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))){
            	// 确保粘滞连接是可用的
                if (availablecheck && stickyInvoker.isAvailable()){
                    return stickyInvoker;
                }
            }
        }
        // 使用负载均衡算法挑选出invoker
        Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
        
        if (sticky){
            stickyInvoker = invoker;
        }
        return invoker;
    }
    

    该方法的逻辑是第一次调用先通过负载均衡算法获取服务,若配置了粘滞连接且该服务正常,则在后续调用中一直使用该服务,直到该服务不可用时再使用负载均衡算法获取服务。
    以上就是服务调用前需要执行的保护机制,下面我们就来看看具体的服务调用实现。

    服务调用

    Result result = invoker.invoke是执行服务调用的逻辑,这里的invoker不难理解由具体的协议决定,这里就是DubboInvoker,同样的,DubboInvoker没有该方法的实现,是继承AbstractInvoker的,可以看到Dubbo中大量使用模板方法模式抽离公共的逻辑,因此具体的实现还是在DubboInvoker的doInvoke方法中:

    protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
        inv.setAttachment(Constants.VERSION_KEY, version);
        
        // 获取客户端连接对象
        ExchangeClient currentClient;
        if (clients.length == 1) {
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
        	// 是否异步处理
            boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
            // 是否为单向通信,即只是客户端向服务端发送消息,不需要响应
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
            if (isOneway) {
            	// 单向通信,因为不需要等待返回结果,所以这里应该是是异步的
            	boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
            	// 发送消息
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture(null);
                // 返回空结果
                return new RpcResult();
            } else if (isAsync) {
            	// 异步处理,并将返回结果设置到RPC上下文中,由用户调用ResponseFuture.get获取结果
            	ResponseFuture future = currentClient.request(inv, timeout) ;
                RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
                return new RpcResult();
            } else {
            	// 同步等待结果返回
            	RpcContext.getContext().setFuture(null);
                return (Result) currentClient.request(inv, timeout).get();
            }
        }
    }
    

    该方法就是dubbo协议执行远程调用的逻辑,其中包含了异步同步的处理逻辑,通过ResponseFuture实现,若异步则会将该类的实例设置到上下文环境中,由用户决定何时调用get方法获取结果;若同步则由框架自身调用get方法阻塞等待返回结果。ResponseFuture是一个接口,这里的默认实现是DefaultFuture,先来看看它是如何实现异步处理的:

    // 响应结果
    private volatile Response                     response;
    
    // 接收结果,该方法由HeaderExchangeHandler调用
    public static void received(Channel channel, Response response) {
        try {
            DefaultFuture future = FUTURES.remove(response.getId());
            if (future != null) {
                future.doReceived(response);
            } else {
                logger.warn("The timeout response finally returned at " 
                            + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) 
                            + ", response " + response 
                            + (channel == null ? "" : ", channel: " + channel.getLocalAddress() 
                                + " -> " + channel.getRemoteAddress()));
            }
        } finally {
            CHANNELS.remove(response.getId());
        }
    }
    
    // 处理响应结果
    private void doReceived(Response res) {
        lock.lock();
        try {
        	// 保存响应结果并唤醒阻塞线程(即get方法获取结果)
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
        if (callback != null) {
            invokeCallback(callback);
        }
    }
    
    // 客户端获取响应结果
    public Object get(int timeout) throws RemotingException {
        if (timeout <= 0) {
            timeout = Constants.DEFAULT_TIMEOUT;
        }
        // 还未接收到服务端的响应结果
        if (! isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
            	// 服务端还未响应完成时阻塞线程直到结果返回
                while (! isDone()) {
                    done.await(timeout, TimeUnit.MILLISECONDS);
                    if (isDone() || System.currentTimeMillis() - start > timeout) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
            if (! isDone()) {
                throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
            }
        }
        // 返回结果给用户
        return returnFromResponse();
    }
    
    private Object returnFromResponse() throws RemotingException {
        Response res = response;
        if (res == null) {
            throw new IllegalStateException("response cannot be null");
        }
        if (res.getStatus() == Response.OK) {
            return res.getResult();
        }
        if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
            throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
        }
        throw new RemotingException(channel, res.getErrorMessage());
    }
    

    上面几个方法为处理服务端响应的主要逻辑,用户可以通过该类的get方法获取响应结果,若是还未接受到响应结果就会阻塞线程直到获取到response,所以是异步还是同步就取决于get方法的调用时机。
    我们继续看客户端的请求currentClient.request,但是这个client应该是什么对象呢?
    在这里插入图片描述
    这三个类应该不陌生的,我们在上文分析过,Dubbo默认使用共享连接,因此会使用ReferenceCountExchangeClient对其它两个进行装饰,提供引用计数的功能,这里不过多分析,向下最终都会进入到HeaderExchangeClient的request方法(LazyConnectExchangeClient只是实现懒加载功能,最终也是创建HeaderExchangeClient客户端)并继续调用HeaderExchangeChannel的request方法:

    private final ExchangeChannel channel
    public ResponseFuture request(Object request, int timeout) throws RemotingException {
        return channel.request(request, timeout);
    }
    
    public ResponseFuture request(Object request, int timeout) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        }
        // 封装请求参数对象
        Request req = new Request();
        req.setVersion("2.0.0");
        // 双向通信标识
        req.setTwoWay(true);
        // 请求数据
        req.setData(request);
        DefaultFuture future = new DefaultFuture(channel, req, timeout);
        try{
        	// 使用netty发送消息
            channel.send(req);
        }catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }
    

    这里逻辑很清晰,也验证了上文的ResponseFuture对象实际为DefaultFuture,往下就是通过channel发送请求消息,由于Duboo默认使用的是Netty通信,所以这里应该是NettyChannel对象,
    不过由于参数列表不匹配,这里首先是调用的其父类AbstractPeer的send方法,然后才进入到NettyChannel中:

    // AbstractPeer
    public void send(Object message) throws RemotingException {
        send(message, url.getParameter(Constants.SENT_KEY, false));
    }
    
    // NettyChannel
    public void send(Object message, boolean sent) throws RemotingException {
        super.send(message, sent);
        
        boolean success = true;
        int timeout = 0;
        try {
            ChannelFuture future = channel.write(message);
            if (sent) {
                timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
                success = future.await(timeout);
            }
            Throwable cause = future.getCause();
            if (cause != null) {
                throw cause;
            }
        } catch (Throwable e) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
        }
        
        if(! success) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
                    + "in timeout(" + timeout + "ms) limit");
        }
    }
    

    至此,客户端调用服务的流程就完成了,下面是整个客户端调用的时序图,为简略,省去了一些具体实现类:
    在这里插入图片描述
    由于是Netty通信,数据出站和入站时会有对应的编码和解码操作,这里就不分析了,下面直接看服务端如何接收处理请求的。

    服务端接收请求

    服务端是通过NettyHandler.messageReceived接收处理请求的:

    private final ChannelHandler handler;
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.received(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }
    

    主要逻辑在hander.received方法中,但这个handler有很多实现类,应该是哪一个呢?
    在分析发布服务的时候在HeaderExchanger.bind方法中包装了很多个handler:

    public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }
    

    继续追踪Transporters.bind方法,最终会再对handler进行包装:

    protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                                        .getAdaptiveExtension().dispatch(handler, url)));
    }
    

    这里拿到的扩展点为AllDispatcher,该类同样是包装handler:

    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }
    

    所以,最终的调用处理链为:

    MultiMessageHandler 
    	-> HeartbeatHandler 
    		-> AllChannelHandler
    			-> DecodeHandler
    				-> HeaderExchangeHandler
    

    先来看看各个handler的作用:

    • MultiMessageHandler :复合消息处理
    • HeartbeatHandler :心跳消息的处理
    • AllChannelHandler:线程派发处理器,将接收到的消息封装成ChannelEventRunnable交给线程池处理
    • DecodeHandler:解码处理
    • HeaderExchangeHandler:响应处理

    前两个handler的逻辑没必要分析了,这里从AllChannelHandler开始,这个是Dubbo的线程派发模型的实现。什么是线程派发模型?简单的说就是Dubbo为提高请求处理效率,将能够简单快速处理的逻辑交由IO线程处理,而复杂耗时的操作交由线程池异步处理(哪些操作应放入线程池,可由用户配置决定)。Dubbo共提供了5种线程派发策略:

    • all:所有消息都派发到线程池,包括请求,响应,连接事件,断开事件等
    • direct:所有消息都不派发到线程池,全部在 IO 线程上直接执行
    • message:只有请求和响应消息派发到线程池,其它消息均在 IO 线程上执行
    • execution:只有请求消息派发到线程池,不含响应。其它消息均在 IO 线程上执行
    • connection:在 IO 线程上,将连接断开事件放入队列,有序逐个执行,其它消息派发到线程池

    默认情况下使用的是all,即AllChannelHandler:

    public class AllChannelHandler extends WrappedChannelHandler {
    
    	// 连接消息
        public void connected(Channel channel) throws RemotingException {
            ExecutorService cexecutor = getExecutorService(); 
            try{
                cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.CONNECTED));
            }catch (Throwable t) {
                throw new ExecutionException("connect event", channel, getClass()+" error when process connected event ." , t);
            }
        }
        
        // 连接断开消息
        public void disconnected(Channel channel) throws RemotingException {
            ExecutorService cexecutor = getExecutorService(); 
            try{
                cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.DISCONNECTED));
            }catch (Throwable t) {
                throw new ExecutionException("disconnect event", channel, getClass()+" error when process disconnected event ." , t);
            }
        }
    
    	// 接收响应消息
        public void received(Channel channel, Object message) throws RemotingException {
            ExecutorService cexecutor = getExecutorService();
            try {
                cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
            } catch (Throwable t) {
                throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
            }
        }
    
    	// 异常消息
        public void caught(Channel channel, Throwable exception) throws RemotingException {
            ExecutorService cexecutor = getExecutorService(); 
            try{
                cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.CAUGHT, exception));
            }catch (Throwable t) {
                throw new ExecutionException("caught event", channel, getClass()+" error when process caught event ." , t);
            }
        }
    }
    

    当前流程我们主要看received方法,该方法将请求对象封装到ChannelEventRunnablec类中,而该类是实现了Runnable接口的,意为着我们直接看run方法就可以了:

    public void run() {
        switch (state) {
            case RECEIVED:
                try{
                    handler.received(channel, message);
                }catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                            + ", message is "+ message,e);
                }
                break;
        }
    }
    

    这里我只截取了关键的代码片段,可以看到是直接交给下一个handler处理的,即DecodeHandler处理,该类是执行解码操作,也不分析了,直接看HeaderExchangeHandler.received方法:

    public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {
            if (message instanceof Request) {
                // 处理请求
                Request request = (Request) message;
                if (request.isEvent()) {
                	// 处理事件消息
                    handlerEvent(channel, request);
                } else {
                	// 处理普通请求
                    if (request.isTwoWay()) {
                    	// 双向通信
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                    	// 单向通信,无需返回请求结果
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
            	// 处理响应
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                .....
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }
    
    Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
        Response res = new Response(req.getId(), req.getVersion());
        // 请求错误,返回错误消息
        if (req.isBroken()) {
            Object data = req.getData();
    
            String msg;
            if (data == null) msg = null;
            else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
            else msg = data.toString();
            res.setErrorMessage("Fail to decode request due to: " + msg);
            res.setStatus(Response.BAD_REQUEST);
    
            return res;
        }
        // 调用服务
        Object msg = req.getData();
        try {
            // 继续调用,这里的handler又是哪一个呢?
            Object result = handler.reply(channel, msg);
            res.setStatus(Response.OK);
            res.setResult(result);
        } catch (Throwable e) {
            res.setStatus(Response.SERVICE_ERROR);
            res.setErrorMessage(StringUtils.toString(e));
        }
        return res;
    }
    

    上面交互的逻辑不复杂,执行远程调用时会继续向下调用handler.reply方法,这个handler是什么呢?还记得在DubboProtocol.createServer方法中传入的requestHandler么?

    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
        
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                // 获取服务端Invoker对象
                Invoker<?> invoker = getInvoker(channel, inv);
                //如果是callback 需要处理高版本调用低版本的问题
                if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))){
                    String methodsStr = invoker.getUrl().getParameters().get("methods");
                    boolean hasMethod = false;
                    if (methodsStr == null || methodsStr.indexOf(",") == -1){
                        hasMethod = inv.getMethodName().equals(methodsStr);
                    } else {
                        String[] methods = methodsStr.split(",");
                        for (String method : methods){
                            if (inv.getMethodName().equals(method)){
                                hasMethod = true;
                                break;
                            }
                        }
                    }
                }
                RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                // 调用invoke方法,最终会调用服务端实现的具体服务方法
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
        }
    
    	// 忽略其它方法
    };
    

    通过getInvoker方法获取服务端的Invoker对象,然后通过该对象去调用具体的服务,但这里的Invoker对象具体是哪一个呢?在服务发布章节中,分析过Invoker,是通过JavassistProxyFactory.getInvoker创建了一个AbstractProxyInvoker的匿名内部类,并通过Wrapper去调用代理对象的方法,那这里获取到的就是这个匿名内部类么?
    还没有这么简单,在RegistryProtocol.doLocalExport方法中首先使用InvokerDelegete对Invoker进行了包装,然后调用protocol.export方法,这个之前我们也分析过,会经过ProtocolListenerWrapper -> ProtocolFilterWrapper -> DubboProtocol,主要看看filter.export:

    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
            return protocol.export(invoker);
        }
        return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
    }
    
    private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
        Invoker<T> last = invoker;
        List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
        if (filters.size() > 0) {
            for (int i = filters.size() - 1; i >= 0; i --) {
                final Filter filter = filters.get(i);
                final Invoker<T> next = last;
                last = new Invoker<T>() {
    
                    public Class<T> getInterface() {
                        return invoker.getInterface();
                    }
    
                    public URL getUrl() {
                        return invoker.getUrl();
                    }
    
                    public boolean isAvailable() {
                        return invoker.isAvailable();
                    }
    
                    public Result invoke(Invocation invocation) throws RpcException {
                        return filter.invoke(next, invocation);
                    }
    
                    public void destroy() {
                        invoker.destroy();
                    }
    
                    @Override
                    public String toString() {
                        return invoker.toString();
                    }
                };
            }
        }
        return last;
    }
    

    可以看到这里获取了Filter扩展点构建了一个过滤链层层过滤,在配置文件中可以看到有如下过滤器:

    echo=com.alibaba.dubbo.rpc.filter.EchoFilter
    generic=com.alibaba.dubbo.rpc.filter.GenericFilter
    genericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFilter
    token=com.alibaba.dubbo.rpc.filter.TokenFilter
    accesslog=com.alibaba.dubbo.rpc.filter.AccessLogFilter
    activelimit=com.alibaba.dubbo.rpc.filter.ActiveLimitFilter
    classloader=com.alibaba.dubbo.rpc.filter.ClassLoaderFilter
    context=com.alibaba.dubbo.rpc.filter.ContextFilter
    consumercontext=com.alibaba.dubbo.rpc.filter.ConsumerContextFilter
    exception=com.alibaba.dubbo.rpc.filter.ExceptionFilter
    executelimit=com.alibaba.dubbo.rpc.filter.ExecuteLimitFilter
    deprecated=com.alibaba.dubbo.rpc.filter.DeprecatedFilter
    compatible=com.alibaba.dubbo.rpc.filter.CompatibleFilter
    timeout=com.alibaba.dubbo.rpc.filter.TimeoutFilter
    monitor=com.alibaba.dubbo.monitor.support.MonitorFilter
    validation=com.alibaba.dubbo.validation.filter.ValidationFilter
    cache=com.alibaba.dubbo.cache.filter.CacheFilter
    trace=com.alibaba.dubbo.rpc.protocol.dubbo.filter.TraceFilter
    future=com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter
    

    所以在requestHandler中调用invoker.invoke方法会经过上面的过滤器层层过滤后才到达最后的服务调用。

    总结

    Dubbo的服务发布、订阅以及交互过程是非常复杂的,总共用了三篇文章来理清整个流程,但其中还有很多的细节没有深入讲解,感兴趣的读者可自行分析,下一篇将分析Dubbo的服务目录源码,也是Dubbo系列的最后一篇。

  • 相关阅读:
    mysql和sql server的update语句的区别
    毕业设计进度日志03
    毕业设计进度日志02
    毕业设计进度日志01
    大数据外出实训报告10
    大数据外出实训报告9
    大数据外出实训报告8
    大数据外出实训报告二
    大数据外出实训报告一(2018/7/9)
    软件工程个人作业02
  • 原文地址:https://www.cnblogs.com/yewy/p/13111825.html
Copyright © 2011-2022 走看看