之前我们分析了客户端调用服务端的源码，但是没有涉及到通讯层和序列化层，本文将之前讲过的内容做一次串联。

1.上层通过动态代理调用refer的call，每个refer又对应一个nettyclient，下面来看一下nettyclient的调用服务端操作

private  Response request(Request request,  boolean  async)  throws  TransportException {
     Channel channel =  null ;
 
     Response response =  null ;
 
     try  {
         // return channel or throw exception(timeout or connection_fail)
         channel = borrowObject(); //向连接池拿连接
 
         if  (channel ==  null ) {
             LoggerUtil.error( "NettyClient borrowObject null: url="  + url.getUri() +  " "
                     + MotanFrameworkUtil.toString(request));
             return  null ;
         }
 
         // async request
         response = channel.request(request); //调用channel的request
         // return channel to pool
         returnObject(channel); //归还连接
     }  catch  (Exception e) {
         LoggerUtil.error(
                 "NettyClient request Error: url="  + url.getUri() +  " "  + MotanFrameworkUtil.toString(request), e);
         //TODO 对特定的异常回收channel
         invalidateObject(channel); //销毁坏的连接
 
         if  (e  instanceof  MotanAbstractException) {
             throw  (MotanAbstractException) e;
         }  else  {
             throw  new  MotanServiceException( "NettyClient request Error: url="  + url.getUri() +  " "
                     + MotanFrameworkUtil.toString(request), e);
         }
     }
 
     // aysnc or sync result
     response = asyncResponse(response, async); //处理response
 
     return  response;
}

2.nettychannel的request操作

public  Response request(Request request)  throws  TransportException {
     int  timeout = nettyClient.getUrl().getMethodParameter(request.getMethodName(), request.getParamtersDesc(),
             URLParamType.requestTimeout.getName(), URLParamType.requestTimeout.getIntValue());
     if  (timeout <=  0 ) {
            throw  new  MotanFrameworkException( "NettyClient init Error: timeout("  + timeout +  ") <= 0 is forbid." ,
                    MotanErrorMsgConstant.FRAMEWORK_INIT_ERROR);
        }
     NettyResponseFuture response =  new  NettyResponseFuture(request, timeout,  this .nettyClient); //创建异步response对象
     this .nettyClient.registerCallback(request.getRequestId(), response); //将此response存入到map，处理完后，会移出
     ChannelFuture writeFuture =  this .channel.write(request); //向服务端传递request对象，写之前会进行序列化的操作
 
     boolean  result = writeFuture.awaitUninterruptibly(timeout, TimeUnit.MILLISECONDS); //标识是否成功
 
     if  (result && writeFuture.isSuccess()) {
         response.addListener( new  FutureListener() { //增加response的监听器
             @Override
             public  void  operationComplete(Future future)  throws  Exception {
                 if  (future.isSuccess() || (future.isDone() && ExceptionUtil.isBizException(future.getException()))) {
                     // 成功的调用
                     nettyClient.resetErrorCount(); //成功
                 }  else  {
                     // 失败的调用
                     nettyClient.incrErrorCount(); //对失败次数+1，如果同一个client连续失败达到所有的连接次数时，标识此client不可用，由心跳管理器负责恢复此client的可用状态
                 }
             }
         });
         return  response; //返回此response,此response为异步的response，由业务线程接手后续接收的过程
     }
 
     writeFuture.cancel();
     response =  this .nettyClient.removeCallback(request.getRequestId()); //在map中移出此response
 
     if  (response !=  null ) {
         response.cancel();
     }
 
     // 失败的调用
     nettyClient.incrErrorCount();
 
     if  (writeFuture.getCause() !=  null ) {
         throw  new  MotanServiceException( "NettyChannel send request to server Error: url="
                 + nettyClient.getUrl().getUri() +  " local="  + localAddress +  " "
                 + MotanFrameworkUtil.toString(request), writeFuture.getCause());
     }  else  {
         throw  new  MotanServiceException( "NettyChannel send request to server Timeout: url="
                 + nettyClient.getUrl().getUri() +  " local="  + localAddress +  " "
                 + MotanFrameworkUtil.toString(request));
     }
}

3.异步的response NettyResponseFuture

public  Object getValue() {
     synchronized  (lock) {
         if  (!isDoing()) {
             return  getValueOrThrowable(); //返回成功值或失败
         }
 
         if  (timeout <=  0 ) {
             try  {
                 lock.wait(); //未接收完毕则一直等待
             }  catch  (Exception e) {
                 cancel( new  MotanServiceException( "NettyResponseFuture getValue InterruptedException : "
                         + MotanFrameworkUtil.toString(request) +  " cost="
                         + (System.currentTimeMillis() - createTime), e));
             }
 
             // don't need to notifylisteners, because onSuccess or
             // onFailure or cancel method already call notifylisteners
             return  getValueOrThrowable();
         }  else  {
             long  waitTime = timeout - (System.currentTimeMillis() - createTime); //等待的时间
 
             if  (waitTime >  0 ) {
                 for  (;;) {
                     try  {
                         lock.wait(waitTime); //要么被通知，要么超时
                     }  catch  (InterruptedException e) {
                     }
 
                     if  (!isDoing()) {
                         break ;
                     }  else  {
                         waitTime = timeout - (System.currentTimeMillis() - createTime);
                         if  (waitTime <=  0 ) {
                             break ;
                         }
                     }
                 }
             }
 
             if  (isDoing()) {
                 timeoutSoCancel();
             }
         }
         return  getValueOrThrowable();
     }
}

本章知识点：

1.motan通过NettyResponseFuture来实现在框架层面异步处理同一笔业务，提升了框架的性能；

2.对于连续失败的client，进行下线操作。