dubbo源码分析系列(4)dubbo通信设计

1 系列目录

2 NIO通信层的抽象

目前dubbo已经集成的有netty、mina、grizzly。先来通过案例简单了解下netty、mina编程(grizzly没有了解过)

2.1 netty和mina的简单案例

netty原本是jboss开发的,后来单独出来了,所以会有两种版本就是org.jboss.netty和io.netty两种包类型的,而dubbo内置的是前者。目前还不是很熟悉,可能稍有差别,但是整体大概都是一样的。

我们先来看下io.netty的案例:

public static void main(String[] args){
    EventLoopGroup bossGroup=new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap serverBootstrap=new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TcpServerHandler());
                }
            });
        ChannelFuture f=serverBootstrap.bind(8080).sync();
        f.channel().closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }finally {  
        workerGroup.shutdownGracefully();  
        bossGroup.shutdownGracefully();  
    }  
}

mina的案例:

public static void main(String[] args) throws IOException{
    IoAcceptor acceptor = new NioSocketAcceptor();
    acceptor.getFilterChain().addLast("codec",new ProtocolCodecFilter(
            new TextLineCodecFactory(Charset.forName("UTF-8"),"\r\n", "\r\n")));
    acceptor.setHandler(new TcpServerHandler());  
    acceptor.bind(new InetSocketAddress(8080));  
}

两者都是使用Reactor模型结构。而最原始BIO模型如下:

dubbo源码分析系列(4)dubbo通信设计

每来一个Socket连接都为该Socket创建一个线程来处理。由于总线程数有限制,导致Socket连接受阻,所以BIO模型并发量并不大

Rector多线程模型如下,更多信息见Netty系列之Netty线程模型

dubbo源码分析系列(4)dubbo通信设计

用一个boss线程,创建Selector,用于不断监听Socket连接、客户端的读写操作等

用一个线程池即workers,负责处理Selector派发的读写操作。

由于boss线程可以接收更多的Socket连接,同时可以充分利用线程池中的每个线程,减少了BIO模型下每个线程为单独的socket的等待时间。

2.2 服务器端如何集成netty和mina

先来简单总结下上述netty和mina的相似之处,然后进行抽象概括成接口

  • 1 各自有各自的编程启动方式
  • 2 都需要各自的ChannelHandler实现,用于处理各自的Channel或者IoSession的连接、读写等事件。

    对于netty来说: 需要继承org.jboss.netty.channel.SimpleChannelHandler(或者其他方式),来处理org.jboss.netty.channel.Channel的连接读写事件

    对于mina来说:需要继承org.apache.mina.common.IoHandlerAdapter(或者其他方式),来处理org.apache.mina.common.IoSession的连接读写事件

为了统一上述问题,dubbo需要做如下事情:

  • 1 定义dubbo的com.alibaba.dubbo.remoting.Channel接口

    • 1.1 针对netty,上述Channel的实现为NettyChannel,内部含有一个netty自己的org.jboss.netty.channel.Channel channel对象,即该com.alibaba.dubbo.remoting.Channel接口的功能实现全部委托为底层的org.jboss.netty.channel.Channel channel对象来实现

    • 1.2 针对mina,上述Channel实现为MinaChannel,内部包含一个mina自己的org.apache.mina.common.IoSession session对象,即该com.alibaba.dubbo.remoting.Channel接口的功能实现全部委托为底层的org.apache.mina.common.IoSession session对象来实现

  • 2 定义自己的com.alibaba.dubbo.remoting.ChannelHandler接口,用于处理com.alibaba.dubbo.remoting.Channel接口的连接读写事件,如下所示

    public interface ChannelHandler {

       void connected(Channel channel) throws RemotingException;
    
       void disconnected(Channel channel) throws RemotingException;
    
       void sent(Channel channel, Object message) throws RemotingException;
    
       void received(Channel channel, Object message) throws RemotingException;
    
       void caught(Channel channel, Throwable exception) throws RemotingException;
    

    }

    • 2.1 先定义用于处理netty的NettyHandler,需要按照netty的方式继承netty的org.jboss.netty.channel.SimpleChannelHandler,此时NettyHandler就可以委托dubbo的com.alibaba.dubbo.remoting.ChannelHandler接口实现来完成具体的功能,在交给com.alibaba.dubbo.remoting.ChannelHandler接口实现之前,需要先将netty自己的org.jboss.netty.channel.Channel channel转化成上述的NettyChannel,见NettyHandler

      public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {

        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            if (channel != null) {
                channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
            }
            handler.connected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
      

      }

      @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {

        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            channels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()));
            handler.disconnected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
      

      }

    • 2.2 先定义用于处理mina的MinaHandler,需要按照mina的方式继承mina的org.apache.mina.common.IoHandlerAdapter,此时MinaHandler就可以委托dubbo的com.alibaba.dubbo.remoting.ChannelHandler接口实现来完成具体的功能,在交给com.alibaba.dubbo.remoting.ChannelHandler接口实现之前,需要先将mina自己的org.apache.mina.common.IoSession转化成上述的MinaChannel,见MinaHandler

      public void sessionOpened(IoSession session) throws Exception {

        MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
        try {
            handler.connected(channel);
        } finally {
            MinaChannel.removeChannelIfDisconnectd(session);
        }
      

      }

      @Override public void sessionClosed(IoSession session) throws Exception {

        MinaChannel channel = MinaChannel.getOrAddChannel(session, url, handler);
        try {
            handler.disconnected(channel);
        } finally {
            MinaChannel.removeChannelIfDisconnectd(session);
        }
      

      }

做了上述事情之后,全部逻辑就统一到dubbo自己的com.alibaba.dubbo.remoting.ChannelHandler接口如何来处理自己的com.alibaba.dubbo.remoting.Channel接口。

这就需要看下com.alibaba.dubbo.remoting.ChannelHandler接口的实现有哪些:

dubbo源码分析系列(4)dubbo通信设计

  • 3 定义Server接口用于统一大家的启动流程

    先来看下整体的Server接口实现情况

    dubbo源码分析系列(4)dubbo通信设计

如NettyServer的启动流程: 按照netty自己的API启动方式,然后依据外界传递进来的com.alibaba.dubbo.remoting.ChannelHandler接口实现,创建出NettyHandler,最终对用户的连接请求的处理全部交给NettyHandler来处理,NettyHandler又交给了外界传递进来的com.alibaba.dubbo.remoting.ChannelHandler接口实现。

至此就将所有底层不同的通信实现全部转化到了外界传递进来的com.alibaba.dubbo.remoting.ChannelHandler接口的实现上了。

而上述Server接口的另一个分支实现HeaderExchangeServer则充当一个装饰器的角色,为所有的Server实现增添了如下功能:

向该Server所有的Channel依次进行心跳检测:

-    如果当前时间减去最后的读取时间大于heartbeat时间或者当前时间减去最后的写时间大于heartbeat时间,则向该Channel发送一次心跳检测
-    如果当前时间减去最后的读取时间大于heartbeatTimeout,则服务器端要关闭该Channel,如果是客户端的话则进行重新连接(客户端也会使用这个心跳检测任务)

2.3 客户端如何集成netty和mina

服务器端了解了之后,客户端就也非常清楚了,整体类图如下:

dubbo源码分析系列(4)dubbo通信设计

如NettyClient在使用netty的API开启客户端之后,仍然使用NettyHandler来处理。还是最终转化成com.alibaba.dubbo.remoting.ChannelHandler接口实现上了。

HeaderExchangeClient和上面的HeaderExchangeServer非常类似,就不再提了。

我们可以看到这样集成完成之后,就完全屏蔽了底层通信细节,将逻辑全部交给了com.alibaba.dubbo.remoting.ChannelHandler接口的实现上了。从上面我们也可以看到,该接口实现也会经过层层装饰类的包装,才会最终交给底层通信。

如HeartbeatHandler装饰类:

public void sent(Channel channel, Object message) throws RemotingException {
    setWriteTimestamp(channel);
    handler.sent(channel, message);
}

public void received(Channel channel, Object message) throws RemotingException {
    setReadTimestamp(channel);
    if (isHeartbeatRequest(message)) {
        Request req = (Request) message;
        if (req.isTwoWay()) {
            Response res = new Response(req.getId(), req.getVersion());
            res.setEvent(Response.HEARTBEAT_EVENT);
            channel.send(res);
            if (logger.isInfoEnabled()) {
                int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
                if(logger.isDebugEnabled()) {
                    logger.debug("Received heartbeat from remote channel " + channel.getRemoteAddress()
                                    + ", cause: The channel has no data-transmission exceeds a heartbeat period"
                                    + (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
                }
            }
        }
        return;
    }
    if (isHeartbeatResponse(message)) {
        if (logger.isDebugEnabled()) {
            logger.debug(
                new StringBuilder(32)
                    .append("Receive heartbeat response in thread ")
                    .append(Thread.currentThread().getName())
                    .toString());
        }
        return;
    }
    handler.received(channel, message);
}

就会拦截那些上述提到的心跳检测请求。更新该Channel的最后读写时间。

2.4 同步调用和异步调用的实现

首先设想一下我们目前的通信方式,使用netty mina等异步事件驱动的通信框架,将Channel中信息都分发到Handler中去处理了,Handler中的send方法只负责不断的发送消息,receive方法只负责不断接收消息,这时候就产生一个问题:

客户端如何对应同一个Channel的接收的消息和发送的消息之间的匹配呢?

这也很简单,就需要在发送消息的时候,必须要产生一个请求id,将调用的信息连同id一起发给服务器端,服务器端处理完毕后,再将响应信息和上述请求id一起发给客户端,这样的话客户端在接收到响应之后就可以根据id来判断是针对哪次请求的响应结果了。

来看下DubboInvoker中的具体实现

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) {
    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();
}
  • 如果不需要返回值,直接使用send方法,发送出去,设置当期和线程绑定RpcContext的future为null
  • 如果需要异步通信,使用request方法构建一个ResponseFuture,然后设置到和线程绑定RpcContext中
  • 如果需要同步通信,使用request方法构建一个ResponseFuture,阻塞等待请求完成

可以看到的是它把ResponseFuture设置到与当前线程绑定的RpcContext中了,如果我们要获取异步结果,则需要通过RpcContext来获取当前线程绑定的RpcContext,然后就可以获取Future对象。如下所示:

String result1 = helloService.hello("World");
System.out.println("result :"+result1);
System.out.println("result : "+RpcContext.getContext().getFuture().get());

当设置成异步请求的时候,result1则为null,然后通过RpcContext来获取相应的值。

然后我们来看下异步请求的整个实现过程,即上述currentClient.request方法的具体内容:

public ResponseFuture request(Object request, int timeout) throws RemotingException {
    // create request.
    Request req = new Request();
    req.setVersion("2.0.0");
    req.setTwoWay(true);
    req.setData(request);
    DefaultFuture future = new DefaultFuture(channel, req, timeout);
    try{
        channel.send(req);
    }catch (RemotingException e) {
        future.cancel();
        throw e;
    }
    return future;
}
  • 第一步:创建出一个request对象,创建过程中就自动产生了requestId,如下

    public class Request {

       private final long    mId;
       private static final AtomicLong INVOKE_ID = new AtomicLong(0);
    
       public Request() {
           mId = newId();
       }
    
       private static long newId() {
           // getAndIncrement()增长到MAX_VALUE时,再增长会变为MIN_VALUE,负数也可以做为ID
           return INVOKE_ID.getAndIncrement();
       }
    

    }

  • 第二步:根据request请求封装成一个DefaultFuture对象,通过该对象的get方法就可以获取到请求结果。该方法会阻塞一直到请求结果产生。同时DefaultFuture对象会被存至DefaultFuture类如下结构中:

    private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();

    key就是请求id

  • 第三步:将上述请求对象发送给服务器端,同时将DefaultFuture对象返给上一层函数,即DubboInvoker中,然后设置到当前线程中

  • 第四步:用户通过RpcContext来获取上述DefaultFuture对象来获取请求结果,会阻塞至服务器端返产生结果给客户端

  • 第五步:服务器端产生结果,返回给客户端会在客户端的handler的receive方法中接收到,接收到之后判别接收的信息是Response后,

    static void handleResponse(Channel channel, Response response) throws RemotingException {

       if (response != null && !response.isHeartbeat()) {
           DefaultFuture.received(channel, response);
       }
    

    }

    就会根据response的id从上述FUTURES结构中查出对应的DefaultFuture对象,并把结果设置进去。此时DefaultFuture的get方法则不再阻塞,返回刚刚设置好的结果。

至此异步通信大致就了解了,但是我们会发现一个问题:

当某个线程多次发送异步请求时,都会将返回的DefaultFuture对象设置到当前线程绑定的RpcContext中,就会造成了覆盖问题,如下调用方式:

String result1 = helloService.hello("World");
String result2 = helloService.hello("java");
System.out.println("result :"+result1);
System.out.println("result :"+result2);
System.out.println("result : "+RpcContext.getContext().getFuture().get());
System.out.println("result : "+RpcContext.getContext().getFuture().get());

即异步调用了hello方法,再次异步调用,则前一次的结果就被冲掉了,则就无法获取前一次的结果了。必须要调用一次就立马将DefaultFuture对象获取走,以免被冲掉。即这样写:

String result1 = helloService.hello("World");
Future<String> result1Future=RpcContext.getContext().getFuture();
String result2 = helloService.hello("java");
Future<String> result2Future=RpcContext.getContext().getFuture();
System.out.println("result :"+result1);
System.out.println("result :"+result2);
System.out.println("result : "+result1Future.get());
System.out.println("result : "+result2Future.get());

最后来张dubbo的解释图片:

dubbo源码分析系列(4)dubbo通信设计

3 通信层与dubbo的结合

从上面可以了解到如何对不同的通信框架进行抽象,屏蔽底层细节,统一将逻辑交给ChannelHandler接口实现来处理。然后我们就来了解下如何与dubbo的业务进行对接,也就是在什么时机来使用上述通信功能:

3.1 服务的发布过程使用通信功能

如DubboProtocol在发布服务的过程中:

  • 1 DubboProtocol中有一个如下结构

    Map<String, ExchangeServer> serverMap

    在发布一个服务的时候会先根据服务的url获取要发布的服务所在的host和port,以此作为key来从上述结构中寻找是否已经有对应的ExchangeServer(上面已经说明)。

  • 2 如果没有的话,则会创建一个,创建过程如下:

    ExchangeServer server = Exchangers.bind(url, requestHandler);

    其中requestHandler就是DubboProtocol自身实现的ChannelHandler。

    获取一个ExchangeServer,它的实现主要是Server的装饰类,依托外部传递的Server来实现Server功能,而自己加入一些额外的功能,如ExchangeServer的实现HeaderExchangeServer,就是加入了心跳检测的功能。

    所以此时我们可以自定义扩展功能来实现Exchanger。接口定义如下:

    @SPI(HeaderExchanger.NAME) public interface Exchanger {

       @Adaptive({Constants.EXCHANGER_KEY})
       ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException;
    
       @Adaptive({Constants.EXCHANGER_KEY})
       ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException;
    

    }

    默认使用的就是HeaderExchanger,它创建的ExchangeServer是HeaderExchangeServer如下所示:

    public class HeaderExchanger implements Exchanger {

       public static final String NAME = "header";
    
       public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
           return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
       }
    
       public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
           return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
       }
    

    }

    HeaderExchangeServer仅仅是一个Server接口的装饰类,需要依托外部传递Server实现来完成具体的功能。此Server实现可以是netty也可以是mina等。所以我们可以自定义Transporter实现来选择不同底层通信框架,接口定义如下:

    @SPI("netty") public interface Transporter {

       @Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})
       Server bind(URL url, ChannelHandler handler) throws RemotingException;
    
       @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
       Client connect(URL url, ChannelHandler handler) throws RemotingException;
    

    } 默认采用netty实现,如下:

    public class NettyTransporter implements Transporter {

       public static final String NAME = "netty";
    
       public Server bind(URL url, ChannelHandler listener) throws RemotingException {
           return new NettyServer(url, listener);
       }
    
       public Client connect(URL url, ChannelHandler listener) throws RemotingException {
           return new NettyClient(url, listener);
       }
    

    } 至此就到了我们上文介绍的内容了。同时DubboProtocol的ChannelHandler实现经过层层装饰器包装,最终传给底层通信Server。

    客户端发送请求给服务器端时,底层通信Server会将请求经过层层处理最终传递给DubboProtocol的ChannelHandler实现,在该实现中,会根据请求参数找到对应的服务器端本地Invoker,然后执行,再将返回结果通过底层通信Server发送给客户端。

3.2 客户端的引用服务使用通信功能

在DubboProtocol引用服务的过程中:

  • 1 使用如下方式创建client

    ExchangeClient client=Exchangers.connect(url ,requestHandler);

    requestHandler还是DubboProtocol中ChannelHandler实现。

    和Server类似,我们可以通过自定义Exchanger实现来创建出不同功能的ExchangeClient。默认的Exchanger实现是HeaderExchanger

    public class HeaderExchanger implements Exchanger {

       public static final String NAME = "header";
    
       public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
           return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
       }
    
       public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
           return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
       }
    

    }

    创建出来的ExchangeClient是HeaderExchangeClient,它也是Client的包装类,仅仅在Client外层加上心跳检测的功能,向它所连接的服务器端发送心跳检测。

    HeaderExchangeClient需要外界给它传一个Client实现,这是由Transporter接口实现来定的,默认是NettyTransporter

    public class NettyTransporter implements Transporter {

       public static final String NAME = "netty";
    
       public Server bind(URL url, ChannelHandler listener) throws RemotingException {
           return new NettyServer(url, listener);
       }
    
       public Client connect(URL url, ChannelHandler listener) throws RemotingException {
           return new NettyClient(url, listener);
       }
    

    }

    创建出来的的Client实现是NettyClient。

    同时DubboProtocol的ChannelHandler实现经过层层装饰器包装,最终传给底层通信Client。

    客户端的DubboInvoker调用远程服务的时候,会将调用信息通过ExchangeClient发送给服务器端,然后返回一个ResponseFuture,根据客户端选择的同步还是异步方式,来决定阻塞还是直接返回,这一部分在上文同步调用和异步调用的实现中已经详细说过了。

4 结束语

本篇文章主要介绍了集成netty和mina的那一块的通信接口及实现的设计,下篇主要介绍编解码的过程

上一篇:ZooKeeper源码研究系列(4)集群版服务器介绍


下一篇:静态网站托管使用阿里云OSS对象存储分分钟搞定