Alogic-FrameWork源碼分析-1.服務(wù)開(kāi)發(fā)

LiveVideoStack 發(fā)布于2019-08-15 10:50 / 1845人閱讀

摘要：介紹是一個(gè)輕量級(jí)的服務(wù)框架，源代碼位于。是的消息路由器，是具體負(fù)責(zé)每一個(gè)請(qǐng)求的路由過(guò)程。當(dāng)執(zhí)行結(jié)束后，向服務(wù)池歸還資源。在多線程模式下，同樣也是在服務(wù)線程中執(zhí)行接口的三種方法。是一個(gè)接口，主要代表服務(wù)輸出的消息實(shí)例。

1.0 Alogic-FrameWork介紹

Alogic-FrameWork是一個(gè)輕量級(jí)的Java服務(wù)框架，源代碼位于Alogic-Github。具有快速開(kāi)發(fā)服務(wù)的特點(diǎn)，在alogic-framework下，一個(gè)成熟的Java開(kāi)發(fā)者可以快速的開(kāi)發(fā)出實(shí)現(xiàn)自己業(yè)務(wù)邏輯的Restful服務(wù)。在這里我們不談具體的開(kāi)發(fā)邏輯，而是專注于分析該框架的源碼。
Alogic-FrameWork的一個(gè)HelloWorld級(jí)別代碼如下：Alogic的HelloWorld-Github
其中主要包括以下幾個(gè)部分：

HelloWorld.java 服務(wù)調(diào)用的具體內(nèi)容

web.xml web項(xiàng)目構(gòu)建的配置文件

settings.xml Alogic服務(wù)目錄配置文件

servant.xml 服務(wù)描述配置文件

1.1 服務(wù)初始化入口

服務(wù)由servlet進(jìn)行攔截，servlet-class對(duì)請(qǐng)求進(jìn)行處理，并返回響應(yīng)。

在一個(gè)HelloWorld級(jí)別的應(yīng)用中，一個(gè)典型的web.xml配置如下：


    MessageRouter
    MessageRouter
    com.anysoft.webloader.ServletAgent
    
        handler
        com.logicbus.backend.server.http.MessageRouterServletHandler
    
    1


    MessageRouter
    /services/*

服務(wù)通過(guò)注冊(cè)ServletAgent類攔截/services/*的全部路由，其中初始化參數(shù)handler為MessageRouterServletHandler類。我們來(lái)看一下如何進(jìn)行這個(gè)過(guò)程：

1.2 服務(wù)上下文的處理

ServletAgent是一個(gè)代理類，繼承自HttpServlet，主要代理了Servlet的初始化、執(zhí)行和銷毀，其中重寫(xiě)了init(ServletConfig servletConfig)，它通過(guò)ClassLoader類加載器加載實(shí)際處理的ServletHandler，并由ServletHandler去執(zhí)行它的初始化方法。

public void init(ServletConfig servletConfig) throws ServletException {
    // 獲取handler參數(shù)
    String handlerClass = servletConfig.getInitParameter("handler");
    // 獲取當(dāng)前Servlet的上下文
    ServletContext sc = servletConfig.getServletContext();
    // 獲取當(dāng)前web項(xiàng)目類加載器
    ClassLoader classLoader = (ClassLoader) sc.getAttribute("classLoader");
    if (classLoader != null) {
        try {
            //創(chuàng)建Handler實(shí)例
            handler = (ServletHandler) classLoader.loadClass(handlerClass)
                    .newInstance();
            //執(zhí)行Handler的初始化方法
            handler.init(servletConfig);
        } catch (Exception e) {
            logger.error("Error occurs when creating handler:"
                    + handlerClass, e);
        }
    } else {
        logger.error("Can not find classLoader");
    }
}

ServletHandler是一個(gè)接口，重新封裝了關(guān)于Servlet的init、service、destory方法；在web.xml中需要配置它的實(shí)現(xiàn)方法，MessageRouterServletHandler是它的具體實(shí)現(xiàn)類，在init實(shí)現(xiàn)方法中，設(shè)置了一些關(guān)鍵屬性，如Http的編碼、跨域、緩存等屬性，以及獲取服務(wù)id、目錄、訪問(wèn)控制等。

在doService實(shí)現(xiàn)方法中，將上述初始化的一些列屬性設(shè)置，達(dá)到重新封裝Http請(qǐng)求的目的；service方法初始化Context，并將參數(shù)作為輸入傳入到action方法中。如下：

    // 初始化HttpContext,HttpContext是Context類的一個(gè)實(shí)現(xiàn)，它是一個(gè)封裝后的Http請(qǐng)求的上下文。
    HttpContext ctx = new HttpContext(request,response,encoding,interceptMode);
    // 獲取當(dāng)前服務(wù)路徑的id
    Path id = normalizer.normalize(ctx, request);
    MessageRouter.action(id,ctx,ac);

1.3 服務(wù)請(qǐng)求過(guò)程

上文提到過(guò)，MessageRouterServletHandler實(shí)現(xiàn)了ServletHandler接口，將Http上下文封裝起來(lái)，同時(shí)將doService方法中獲取得到的服務(wù)id，訪問(wèn)控制給MessageRouter的action方法。MessageRouter是Alogic的消息路由器，是具體負(fù)責(zé)每一個(gè)請(qǐng)求的路由過(guò)程。在action方法中，包括以下邏輯：

處理路由追蹤

獲取服務(wù)實(shí)例池

通過(guò)訪問(wèn)控制器分配訪問(wèn)優(yōu)先級(jí)

從服務(wù)實(shí)例池獲取實(shí)例

日志記錄

首先，MessageRouter根據(jù)獲取得到的服務(wù)id來(lái)獲取一個(gè)服務(wù)實(shí)例池，通過(guò)資源池模式來(lái)保證服務(wù)實(shí)例的不斷重復(fù)利用。資源池獲取代碼如下：

        // 獲取服務(wù)實(shí)例池
        ServantFactory factory = servantFactory;
        // 根據(jù)服務(wù)id獲取服務(wù)工廠
        pool = factory.getPool(id);        
        if (!pool.isRunning()){
            throw new ServantException("core.service_paused",
                "The Service is paused:service id:" + id);
        }

而接著，對(duì)于已經(jīng)獲得的資源池中根據(jù)優(yōu)先級(jí)獲得服務(wù)實(shí)例。在非線程模式下調(diào)用execute()方法，在多線程模式下建立服務(wù)工作線程。當(dāng)執(zhí)行結(jié)束后，向服務(wù)池歸還資源。

        //從服務(wù)實(shí)例池中拿服務(wù)實(shí)例，并執(zhí)行
        servant = pool.borrowObject(priority);
        // 判斷是否獲取到了服務(wù)并輸出錯(cuò)誤日志
        if (servant == null){
            logger.warn("Can not get a servant from pool in the limited time,check servant.queueTimeout variable.");
            ctx.setReturn("core.time_out", "Can not get a servant from pool in the limited time,check servant.queueTimeout variable.");
        }else{
               if (!threadMode){
                //在非線程模式下,不支持服務(wù)超時(shí)
                execute(servant,ctx);
            }else{
                // 構(gòu)建CountDownLatch，用于等待服務(wù)工作線程建立。
                CountDownLatch latch = new CountDownLatch(1);
                //建立服務(wù)工作線程
                ServantWorkerThread thread = new ServantWorkerThread(servant,ctx,latch,tc != null ? tc.newChild() : null);
                thread.start();
                // 判斷服務(wù)工作線程是否在指定的時(shí)間內(nèi)建立完成。如果超時(shí)則取消主線程阻塞狀態(tài)，并
                if (!latch.await(servant.getTimeOutValue(), TimeUnit.MILLISECONDS)){
                    ctx.setReturn("core.time_out","Time out or interrupted.");
                }
                thread = null;
            }
        }
    }catch (ServantException ex){
        ctx.setReturn(ex.getCode(), ex.getMessage());
        logger.error(ex.getCode() + ":" + ex.getMessage());
    }catch (Exception ex){
        ctx.setReturn("core.fatalerror",ex.getMessage());
        logger.error("core.fatalerror:" + ex.getMessage(),ex);
    }catch (Throwable t){
        ctx.setReturn("core.fatalerror",t.getMessage());
        logger.error("core.fatalerror:" + t.getMessage(),t);            
    }
    finally {
            ctx.setEndTime(System.currentTimeMillis());
            if (ctx != null){
                // 完成Context
                ctx.finish();
            }
            if (pool != null){
                if (servant != null){
                    // 向服務(wù)池歸還資源
                    pool.returnObject(servant);        
                }
                // 服務(wù)池訪問(wèn)一次                
                pool.visited(ctx.getDuration(),ctx.getReturnCode());
                if (ac != null){
                    ac.accessEnd(sessionId,id, pool.getDescription(), ctx);
                }                
            }                        
            if (bizLogger != null){                
                //需要記錄日志
                log(id,sessionId,pool == null ? null : pool.getDescription(),ctx);
            }
            if (tracerEnable){
                boolean ok = ctx.getReturnCode().equals("core.ok");
                Tool.end(tc, "ALOGIC", id.getPath(), ok ?"OK":"FAILED", ok ? ctx.getQueryString() : ctx.getReason(), ctx.getContentLength());
            }
        }

在非線程模式下的execute方法執(zhí)行了服務(wù)調(diào)用的前置方法、執(zhí)行方法和后置方法。

    protected static int execute(Servant servant,Context ctx) throws Exception {
        servant.actionBefore(ctx);
        servant.actionProcess(ctx);
        servant.actionAfter(ctx);
        return 0;
    }

在多線程模式下，同樣也是在服務(wù)線程中執(zhí)行Servant接口的三種方法。

    public void run(){
        TraceContext tc = null;
        if (traceCtx != null){
            tc = Tool.start(traceCtx.sn(), traceCtx.order());
        }
        boolean error = false;
        try
        {
            m_servant.actionBefore(m_ctx);
            m_servant.actionProcess(m_ctx);
            m_servant.actionAfter(m_ctx);
        }
    }

1.4 服務(wù)響應(yīng)

在MessageRouter的acion方法中，服務(wù)調(diào)用的最后會(huì)調(diào)用ctx.finish()，在這個(gè)方法中調(diào)用了msg的finish方法。

    try {
            if (!isIgnore()){
                if (msg == null){
                    if (getReturnCode().equals("core.ok")){
                        response.sendError(404, "No message is found,check servant implemention.");
                    }else{
                        response.sendError(404, getReturnCode() + ":" + getReason());
                    }
                }else{
                    response.setCharacterEncoding(encoding);
                    msg.finish(this,!cometMode());
                }
            }
        }

Message是一個(gè)接口，主要代表服務(wù)輸出的消息實(shí)例。在Alogic中，Message可以有XML、JSON等協(xié)議的消息實(shí)例，如輸出為RawMessage時(shí)，finish方法如下：

    public void finish(Context ctx,boolean closeStream) {
        // 設(shè)置輸出流
        OutputStream out = null;
        try {
            // 設(shè)置返回內(nèi)容格式
            ctx.setResponseContentType(contentType);
            out = ctx.getOutputStream();
            byte [] bytes = buf.toString().getBytes(ctx.getEncoding());
            contentLength += bytes.length;
            // 將字符串寫(xiě)到輸出流中
            Context.writeToOutpuStream(out, bytes);
            // 輸出打印
            out.flush();
        }catch (Exception ex){
            logger.error("Error when writing data to outputstream",ex);
        }finally{
            if (closeStream)
            IOTools.close(out);
        }
    }

到此，一個(gè)服務(wù)的執(zhí)行邏輯如下：