Dubbo通信模型

本文隶属于分类

互联网

广告推荐

技术交流学习或者有任何问题欢迎加群

编程技术交流群 : 154514123 爱上编程      Java技术交流群 : 6128790  Java

Dubbo和通信结合

通信实现

服务的发布过程使用通信功能:
Protocol.export()时会为每个服务创建一个Server

服务的引用过程使用通信功能:
Protocol.refer()时会创建一个Client

整个类结构及调用关系如下:

这里写图片描述

从图中可以看出,Dubbo的Transporter层完成通信功能,底层的Netty和Mina委托给统一的ChannelHandler来完成具体的功能

编解码

Socket是对TCP/IP的封装和应用,TCP/IP都有一个报文头结构定义,作用非常大,例如解决粘包问题。Dubbo借助Netty已经将这样一部分工作委托出去了,不过还是有些工作需要Dubbo来完成,我们来看一张官方提供的报文头定义:
这里写图片描述
只有搞清楚了报文头定义,才能完成报文体的编码解码,交给底层通信框架去收发

序列化

Dubbo本身支持多种序列化方式,具体使用哪种序列化方式需要由业务场景来决定,详见Dubbo官网

NIO通信层

Dubbo已经集成的有Netty、Mina,重点分析下Netty,详见Netty系列之Netty线程模型

服务器端

NettyServer的启动流程: 首先创建出NettyHandler,用户的连接请求的处理全部交给NettyHandler来处理,NettyHandler又会委托ChannelHandler接口做Dubbo具体的事情。

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

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

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

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

看下ChannelHandler接口的实现情况:

这里写图片描述

看下Server接口实现情况:

这里写图片描述

客户端

看下Client接口实现情况:
这里写图片描述

NettyClient在使用Netty的API开启客户端之后,仍然使用NettyHandler来处理,还是最终委托给ChannelHandler接口实现上

我们可以发现,这样集成完成之后,就完全屏蔽了底层通信细节,将逻辑全部交给了ChannelHandler

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

该部分主要在Client端,调用过程DubboProtocol.refer()->DubboInvoker,

来看下DubboInvoker的具体实现:

    @Override
    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) {
                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();
            }
        } catch (TimeoutException e) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (RemotingException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }
  1. 如果不需要返回值,直接使用send方法,发送出去,设置当期和线程绑定RpcContext的future为null
  2. 如果需要异步通信,使用request方法构建一个ResponseFuture,然后设置到和线程绑定的RpcContext中
  3. 如果需要同步通信,使用request方法构建一个ResponseFuture,阻塞等待请求完成

另外官方文档有说明(Dubbo协议):Dubbo协议采用单一长连接和NIO异步通讯(默认Netty,Netty使用Socket(通信是全双工的方式,可以更方便的使用TCP/IP协议栈)完成通信)
适合于小数据量大并发的服务调用,以及服务消费者机器数远大于服务提供者机器数的情况

Dubbo协议线程说明

这里写图片描述

Dubbo协议:

  • 连接个数:单连接
  • 连接方式:长连接
  • 传输协议:TCP
  • 传输方式:NIO异步传输
  • 序列化:Hessian
  • 适用范围:入传出参数数据包较小(建议小于100K),消费者比提供者个数多,单一消费者无法压满提供者,尽量不要用dubbo协议传输大文件或超大字符串

同步调用

我们首先看第3种情况,对于当前线程来说,将请求发送出去,暂停等结果回来后再执行。于是这里出现了2个问题:

  • 当前线程怎么让它“暂停,等结果回来后,再执行?
  • 正如前面所说,Socket通信是一个全双工的方式,如果有多个线程同时进行远程方法调用,这时建立在client server之间的socket连接上会有很多双方发送的消息传递,前后顺序也可能是乱七八糟的,server处理完结果后,将结果消息发送给client,client收到很多消息,怎么知道哪个消息结果是原先哪个线程调用的?

我们从代码上找些痕迹
调用路径:HeaderExchangeClient.request()->HeaderExchangeChannel.request()

    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!");
        }
        // 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;
    }

注意这个方法返回的ResponseFuture对象,当前客户端请求的线程在经过一系列调用后,会拿到ResponseFuture对象,最终该线程会阻塞在这个对象的下面这个方法调用上,如下:

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

上面我已经看到请求线程已经阻塞,那么又是如何被唤醒的呢?
上文提到过Client端的处理最终转化成ChannelHandler接口实现上,我们看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) {
                // handle 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) {
            //这里就是作为消费者的dubbo客户端在接收到响应后,触发通知对应等待线程的起点
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                if (isClientSide(channel)) {
                    Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                    logger.error(e.getMessage(), e);
                } else {
                    String echo = handler.telnet(channel, (String) message);
                    if (echo != null && echo.length() > 0) {
                        channel.send(echo);
                    }
                }
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }

    static void handleResponse(Channel channel, Response response) throws RemotingException {
        if (response != null && !response.isHeartbeat()) {
            DefaultFuture.received(channel, response);
        }
    }

熟悉的身影:DefaultFuture,继续看received()方法

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

留一下我们之前提到的id的作用,这里可以看到它已经开始发挥作用了。通过id,DefaultFuture.FUTURES可以拿到具体的那个DefaultFuture对象,它就是上面我们提到的,阻塞请求线程的那个对象。好,找到目标后,调用它的doReceived方法,唤醒阻塞的线程,拿到返回结果

    private void doReceived(Response res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
        if (callback != null) {
            invokeCallback(callback);
        }
    }

现在前面2个问题已经有答案了

  • 当前线程怎么让它“暂停”,等结果回来后,再向后执行?
    答:先生成一个对象ResponseFuture,在一个全局map里put(ID,Future)存放起来,使用ResponseFuture的ReentrantLock.lock()让当前线程处于等待状态,然后另一消息监听线程等到服务端结果来了后,再map.get(ID)找到ResponseFuture,调用ResponseFuture.unlock()唤醒前面处于等待状态的线程。

  • 正如前面所说,Socket通信是一个全双工的方式,如果有多个线程同时进行远程方法调用,这时建立在client server之间的socket连接上会有很多双方发送的消息传递,前后顺序也可能是乱七八糟的,server处理完结果后,将结果消息发送给client,client收到很多消息,怎么知道哪个消息结果是原先哪个线程调用的?
    答:使用一个ID,让其唯一,然后传递给服务端,再服务端又回传回来,这样就知道结果是原先哪个线程的了。

异步调用

官方给出了异步调用的文档
异步调用先返回一个ResponseFuture对象,然后设置到和线程绑定的RpcContext中去

此时我们会发现一个问题,当某个线程多次发送异步请求时,都会将返回的DefaultFuture对象设置到当前线程绑定(ThreadLocal是个静态常量)的RpcContext中,就会造成了覆盖问题,如下调用方式:

//RpcContext.getContext().setFuture()
String result1 = helloService.hello("World");
//RpcContext.getContext().setFuture()
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());

技术交流学习或者有任何问题欢迎加群

编程技术交流群 : 154514123 爱上编程      Java技术交流群 : 6128790  Java

广告推荐

讨论区