Gaea源碼閱讀（二）：用戶端流程

以GaeaClientTest為入口

GaeaInit.init("conf/gaea.config");
		/**
		 * 調用URL 格式:tcp://服務名//接口實作類 
		 * 備注: 
		 * 服務名：需要與gaea.config中的服務名一一對應
		 * 接口實作類：具體調用接口的接口實作類
		 */

		final String url = "tcp://demo/NewsService";
		INewsService newsService = ProxyFactory.create(INewsService.class, url);
		List<News> list = newsService.getNewsByCateID();
		for (News news : list) {
			System.out.println("ID is " + news.getNewsID() + " title is "
					+ news.getTitle());
		}
           

如注釋所言，用戶端通過Restfull格式url"tcp://serviceName/lookup "通路服務。

關鍵的一句：INewsService newsService = ProxyFactory. create(INewsService. class,url); ，這句話建立了一個實作了INewsService接口的代理，傳回的這個代理對象将作為getNewsByCateID的調用句柄。

讓我們看看代理的内部如何實作的，跟蹤ProxyFactory.create

InvocationHandler handler = new ProxyStandard(type,serviceName, lookup);
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                newClass[]{type},
               handler);
 
       //代理的作用就是将所有調用轉交到Handler
           

InvocationHandler接口實作類ProxyStandard

public ProxyStandard(Class<?> interfaceClass, String serviceName, String lookup){
        this.interfaceClass =interfaceClass;
        this.methodCaller = newMethodCaller(serviceName, lookup);
}
           

InvocationHandler主要由methodCaller完成功能，在invoke()中交給methodCaller

        Object obj= methodCaller.doMethodCall(args, method);

MethodCaller機制 //doMethodCall

//準備參數
        Type[]typeAry = methodInfo.getGenericParameterTypes();
        Class<?>[] clsAry = methodInfo.getParameterTypes();
        if (args== null) {
            args = newObject[0];
        }
 
        Parameter[]paras = new Parameter[args.length];
        List<Integer> outParas = newArrayList<Integer>();
 
        if(typeAry != null) {
            for (int i = 0;i < typeAry.length; i++) {
                if(args[i] instanceof Out) {
                   paras[i] = newParameter(args[i], clsAry[i], typeAry[i], ParaType.Out);
                    outParas.add(i);
                } else {
                   paras[i] = newParameter(args[i], clsAry[i], typeAry[i], ParaType.In);
                }
            }
        }
 
        //規範化方法名
        StringmethodName = methodInfo.getName();
        OperationContract ann =methodInfo.getAnnotation(OperationContract.class);
        if (ann !=null) {
            if(!ann.methodName().equals(AnnotationUtil.DEFAULT_VALUE)) {
               methodName = "$" + ann.methodName();
            }
        }
 
        ParameterreturnPara = new Parameter(null,methodInfo.getReturnType(), methodInfo.getGenericReturnType());
 
        //調用方法
        ServiceProxy proxy = ServiceProxy.getProxy(serviceName);
 
        InvokeResultresult = proxy.invoke(returnPara, lookup,methodName, paras);
 
        //設定傳回值
        if (result!= null && result.getOutPara() != null) {
            for (int i = 0;i < outParas.size() && i < result.getOutPara().length; i++) {
               Object op = args[outParas.get(i)];
                if(op instanceof Out){
                   ((Out)op).setOutPara(result.getOutPara()[i]);
                }
            }
        }
        returnresult.getResult();
           

這裡出現了OperationContract注解，這是在方法上面注解的。注解有一個methodName屬性。

關鍵點是ServiceProxy的getProxy和invoke方法。

ServiceProxy.getProxy：構造ServiceProxy

config =ServiceConfig.GetConfig(serviceName);
        dispatcher = newDispatcher(config);
       
        requestTime = config.getSocketPool().getReconnectTime();
             intserverCount = 1;
             if(dispatcher.GetAllServer()!= null && dispatcher.GetAllServer().size()> 0){
                       serverCount = dispatcher.GetAllServer().size();
             }
            
             ioreconnect =serverCount - 1;
 
         //     count = max {ioreconnect,requestTime}
             count = requestTime;
            
             if(ioreconnect > requestTime){
                       count = ioreconnect;
             }
           

                   首先讀取配置檔案中屬性為//Service[@name= serviceName]的節點

                   建立實作負載均衡的dispatcher，在這裡建立了Server

for(ServerProfile ser : config.getServers()) {
            if(ser.getWeithtRate() > 0) {
               Server s = newServer(ser);
                if(s.getState() != ServerState.Disable) {
                   ScoketPool sp = newScoketPool(s, config);
                   s.setScoketpool(sp);
                   ServerPool.add(s);
                }
            }
        }
 
           

ServiceProxy.Invoke方法

//構造RequestProtocol
        RequestProtocol requestProtocol = newRequestProtocol(typeName, methodName, listPara);
        ProtocolsendP = new Protocol(createSessionId(),
                           (byte) config.getServiceid(),
                           SDPType.Request,
                           CompressType.UnCompress,
                           config.getProtocol().getSerializerType(),
                           PlatformType.Java,
                           requestProtocol);
       
        ProtocolreceiveP = null;
        Serverserver = null;
       
        for(int i = 0;i <= count; i++){
                 server = dispatcher.GetServer();
            if (server== null) {
                logger.error("cannotget server");
                throw newException("cannot get server");
            }
            try{
                     //本地存根調用
                     receiveP = server.request(sendP);
                     break;
            } catch(IOExceptionio){
                    
            } catch(RebootExceptionrb){
                     this.createReboot(server);
            }catch(TimeoutExceptionte){
                    
            } catch(Throwable ex){
                    
            }
        }
       
        if(receiveP== null){
                 throw newException("userdatatype error!");
        }
       
        if(receiveP.getSDPType() == SDPType.Response) {
            ResponseProtocol rp = (ResponseProtocol)receiveP.getSdpEntity();
            logger.debug("invoketime:" + (System.currentTimeMillis() - watcher) + "ms");
            return new InvokeResult(rp.getResult(),rp.getOutpara());
        } else if(receiveP.getSDPType()== SDPType.Reset){
            logger.info(server.getName()+"server is reboot,system will change normal server!");
            this.createReboot(server);
            returninvoke(returnType, typeName, methodName, paras);
        }else if(receiveP.getSDPType() == SDPType.Exception) {
            ExceptionProtocol ep = (ExceptionProtocol)receiveP.getSdpEntity();
            throwThrowErrorHelper.throwServiceError(ep.getErrorCode(), ep.getErrorMsg());
        } else {
            throw newException("userdatatype error!");
        }
           

Server.request()過程

//每個CSocket有個WaitWindows注冊了SessionId到WindowData的表
 
       increaseCU();
        CSocketsocket = null;
        try {
            try {//發送消息
               socket = this.scoketpool.getSocket();
               byte[] data= p.toBytes(socket.isRights(),socket.getDESKey());
               socket.registerRec(p.getSessionID());
               socket.send(data); 
            } catch(Throwable ex) {
                logger.error("Serverget socket Exception", ex);
                throw ex;
            }finally {
                     if(socket!= null){
                               socket.dispose();
                     }
            }
 
            //接收應答
            byte[]buffer = socket.receive(p.getSessionID(), currUserCount);
            Protocol result = Protocol.fromBytes(buffer,socket.isRights(),socket.getDESKey());
            if (this.state ==ServerState.Testing) {
               relive();
            }
            return result;
        } catch(IOException ex) {
           
        } catch(Throwable ex) {
 
        } finally {
          
        }
           

Receive接收應答時，通過sessionId對應的AutoResetEvent等待

//CSocket.receive
        AutoResetEvent event = wd.getEvent();
        int timeout= getReadTimeout(socketConfig.getReceiveTimeout(), queueLen);
        if(!event.waitOne(timeout)) {
            throw newTimeoutException("Receive data timeout or error!timeout:" +timeout + "ms,queue length:" +queueLen);
        }
           

傳回應答可能為Response、Reset、Exception

Response -> 傳回InvokeResult結構體

Reset -> 重新開機則再調用invoke(returnType,typeName, methodName, paras)

Exception -> 抛出異常

frameHandler從Socket接收資料,并根據協定反解出session id,然後從waitWindows中取出對應的WindowData,然後調用對應的Event的set()通知CSocket.receive 傳回

用戶端做的工作主要就是這些