dubbo server处理请求

源码分析基于dubbo 2.6.0

前面讲到，RegistryProtocol.doLocalExport负责启动server端网络通讯服务

    private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
        String key = getCacheKey(originInvoker);
        ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
        if (exporter == null) {
            synchronized (bounds) {
                exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
                if (exporter == null) {
                    // 对Invoker进行了代理
                    final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
                    exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                    bounds.put(key, exporter);
                }
            }
        }
        return exporter;
    }

这里invokerDelegete中url的格式为dubbo://(server ip):(server port)/com.dubbo.start.service.HelloService?class=com.dubbo.start.service.HelloServiceImpl&dubbo=2.6.0&interface=com.dubbo.start.service.HelloService&methods=hello2,hello&...， 所以会调用DubboProtocol，但会先经过两个装饰类ProtocolListenerWrapper，ProtocolFilterWrapper。

先看看DubboProtocol.export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    URL url = invoker.getUrl();

    // 创建DubboExporter
    String key = serviceKey(url);
    DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
    exporterMap.put(key, exporter);

    ...

    openServer(url);
    optimizeSerialization(url);
    return exporter;
}

注意一下，这里会构建DubboExporter，并缓存到exporterMap。（后面会用到）

关键就是openServer

private void openServer(URL url) {
    // find server.
    String key = url.getAddress();
    //client can export a service which's only for server to invoke
    boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
    if (isServer) {
        ExchangeServer server = serverMap.get(key);
        if (server == null) {
            serverMap.put(key, createServer(url));
        } else {
            // server supports reset, use together with override
            server.reset(url);
        }
    }
}

如果ExchangeServer不存在，就createServer

private ExchangeServer createServer(URL url) {
    // url 参数处理
    ...
    ExchangeServer server;
    try {
        server = Exchangers.bind(url, requestHandler);
    } catch (RemotingException e) {
        throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
    }
    ...

    return server;
}

Exchangers.bind会调用HeaderExchanger.bind(注意第二个参数DubboProtocol.requestHandler)

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

出现了一个新的概念Transporter（运输层）。

Transporters同样通过ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension()找到一个Transporter实现类，默认使用netty3， 会根据url中server/transporter参数转换。

我配置了netty4，所以会使用com.alibaba.dubbo.remoting.transport.netty4包下的netty代码。
NettyTransporter

    public Server bind(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyServer(url, listener);
    }

NettyServer构造方法

  public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
        super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
    }

先看看ChannelHandlers.wrap

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

这里出现一个新的概念Dispatcher，它是dubbo中的调度器，默认是AllDispatcher

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

再回调一下HeaderExchanger.bind方法， Transporters.bind(URL url, ChannelHandler... handlers)第二个参数ChannelHandler为new DecodeHandler(new HeaderExchangeHandler(handler)) (注意new HeaderExchangeHandler(handler)中的handler为DubboProtocol.requestHandler)
ChannelHandler是一条责任链，到这里，责任链国家完成，节点依次为 MultiMessageHandler > HeartbeatHandler > AllChannelHandler > DecodeHandler > HeaderExchangeHandler > DubboProtocol.requestHandler

NettyServer继承了AbstractServer，AbstractServer的构造方法会调用NettyServer.doOpen

    protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();

        bootstrap = new ServerBootstrap();
        // bossGroup中线程数为1
        bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
        workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
                new DefaultThreadFactory("NettyServerWorker", true));

        final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
        channels = nettyServerHandler.getChannels();

        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                        ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                                .addLast("decoder", adapter.getDecoder())
                                .addLast("encoder", adapter.getEncoder())
                                .addLast("handler", nettyServerHandler);
                    }
                });
        // bind
        ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
        channelFuture.syncUninterruptibly();
        channel = channelFuture.channel();

    }

关键是netty handler的实现类NettyServerHandler。注意它的构造方法第二个参数为this，它接受的是ChannelHandler类型，就是说NettyServer也继承了ChannelHandler。

NettyServerHandler继承了netty的ChannelDuplexHandler，这里只关注channelRead方法

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
        try {
            handler.received(channel, msg);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.channel());
        }
    }

handler就是NettyServer， 他会调用上面说的ChannelHandler责任链。会调用到AllChannelHandler.received

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) {
        ...
    }
}

就是给ExecutorService提交一个任务，异步处理请求。

看看ChannelEventRunnable.run，会根据state进行不同的处理。但实际请求都转发到handler处理（就是继续调用责任链）。

再看看DecodeHandler.received

    public void received(Channel channel, Object message) throws RemotingException {
        // 解码
        if (message instanceof Decodeable) {
            decode(message);
        }

        if (message instanceof Request) {
            decode(((Request) message).getData());
        }

        if (message instanceof Response) {
            decode(((Response) message).getResult());
        }
        // 调用下一个节点
        handler.received(channel, message);
    }

HeaderExchangeHandler

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);
    }
}

twoWay表示需要响应的请求。这里调用handleRequest方法，将返回一个Response，最后通过channel将它send到client。

Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
    Response res = new Response(req.getId(), req.getVersion());
    
    // find handler by message class.
    Object msg = req.getData();
    try {
        // handle data.
        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;
}

终于调用到DubboProtocol.requestHandler了

public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
    if (message instanceof Invocation) {
        Invocation inv = (Invocation) message;
        Invoker<?> invoker = getInvoker(channel, inv);
        ...
        RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
        return invoker.invoke(inv);
    }
    throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
}

这里也是通过invoker调用，逐渐调用到我们的业务方法。

先看看getInvoker方法

Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
    ...

    DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);

    return exporter.getInvoker();
}

很简单，从exporterMap中获取DubboExporter，再获取对应的Invoker。

这时回顾一下上文说到的DubboProtocol.export方法，它用invoker构建了DubboExporter，缓存到exporterMap中。
invoker也是责任链。那么invoker在哪里创建的呢？

这时要回到dubbo server启动，那里说到ServiceConfig.doExportUrlsFor1Protocol

if (registryURLs != null && registryURLs.size() > 0) {
    for (URL registryURL : registryURLs) {
        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

        Exporter<?> exporter = protocol.export(wrapperInvoker);
    }
}

是的，invoker就是在这里创建的。(注意，参数ref就是方法的实现类，栗子中就是HelloServiceImpl)

proxyFactory默认使用的是JavassistProxyFactory。

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
    // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
    final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
    return new AbstractProxyInvoker<T>(proxy, type, url) {
        @Override
        protected Object doInvoke(T proxy, String methodName,
                                  Class<?>[] parameterTypes,
                                  Object[] arguments) throws Throwable {
            return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
        }
    };
}

前面也说过了，DubboProtocol处理前，会经过两个包装类ProtocolListenerWrapper，ProtocolFilterWrapper。

ProtocolFilterWrapper.export 会调用buildInvokerChain，为每一个filter创建一个invoker，

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 Result invoke(Invocation invocation) throws RpcException {
                    return filter.invoke(next, invocation);
                }
            };
        }
    }
    return last;
}

dubbo默认有如下

• EchoFilter
• ClassLoaderFilter
• GenericFilter
• ContextFilter
• TraceFilter
• TimeoutFilter
• MonitorFilter
• ExceptionFilter
• ValidationFilter

再回顾一下上文的RegistryProtocol.doLocalExport方法，对Invoker进行代理包装，生成了InvokerDelegete。所以invoker责任链节点为ProtocolFilterWrapper&Invoker > RegistryProtocol&InvokerDelegete > DelegateProviderMetaDataInvoker > JavassistProxyFactory&AbstractProxyInvoker。

JavassistProxyFactory中创建的invoke，会通过Wrapper调用到实际的业务方法。

com.alibaba.dubbo.common.bytecode.Wrapper也是动态生成代码，JavassistProxyFactory正是通过它调用业务方法。
看看getWrapper方法

public static Wrapper getWrapper(Class<?> c) {
    while (ClassGenerator.isDynamicClass(c)) // can not wrapper on dynamic class.
        c = c.getSuperclass();

    if (c == Object.class)
        return OBJECT_WRAPPER;

    Wrapper ret = WRAPPER_MAP.get(c);
    if (ret == null) {
        ret = makeWrapper(c);
        WRAPPER_MAP.put(c, ret);
    }
    return ret;
}

关键在makeWrapper，makeWrapper会拼凑代码字符串，再通过javassist生成代理类。
过程比较繁琐，直接看生成的代理类吧。

原接口

public interface HelloService {
    String hello(String user) ;
    String hello2(String user) ;
}

生成的Wrapper，关键的方法在invokeMethod

public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException{ 
    com.dubbo.start.service.HelloService w; 
    try{
        w = ((com.dubbo.start.service.HelloService)$1); 
    } catch(Throwable e){ 
        throw new IllegalArgumentException(e); 
    } 
    try{ 
        if( "hello".equals( $2 )  &&  $3.length == 1 ) {
            return ($w)w.hello((java.lang.String)$4[0]); 
        } 
        if( "hello2".equals( $2 )  &&  $3.length == 1 ) {  
            return ($w)w.hello2((java.lang.String)$4[0]);
        } 
    } catch(Throwable e) {     
        throw new java.lang.reflect.InvocationTargetException(e);  
    }
}

$1, $2, $3在javassist中表示方法参数。
可以看到，通过方法名和参数数调用对应的逻辑方法。
所以dubbo暴露的接口就尽量不要方法重构了