上篇我們了解到服務(wù)啟動(dòng)過(guò)程中，ProviderBootStrap.init()和ProviderBootStrap.startup(providerConfig);分別啟動(dòng)http協(xié)議服務(wù)和default（TCP）協(xié)議的netty服務(wù)。本篇我們分析下端口監(jiān)聽(tīng)和消息處理。

我們以ProviderBootStrap.startup(providerConfig)來(lái)切入TCP協(xié)議NettyServer實(shí)現(xiàn)。
關(guān)鍵代碼如下：

    public static ServerConfig startup(ProviderConfig<?> providerConfig) {
            ServerConfig serverConfig = providerConfig.getServerConfig();
            synchronized (ProviderBootStrap.class) {
                //Server接口兩個(gè)實(shí)現(xiàn)：NettyServer和JettyHttpServer
                List<Server> servers = ExtensionLoader.newExtensionList(Server.class);
                for (Server s : servers) {
                    if (!s.isStarted()) {
                        //NettyServer實(shí)現(xiàn)支持default協(xié)議
                        if (s.support(serverConfig)) {
                            //調(diào)用AbstractServer.start
                            s.start(serverConfig);
                            s.addService(providerConfig);
                            serversMap.put(s.getProtocol() + serverConfig.getPort(), s);
                            logger.warn("pigeon " + s + "[version:" + VersionUtils.VERSION + "] has been started");
                            break;
                        }
                    }
                }
                server = serversMap.get(serverConfig.getProtocol() + serverConfig.getPort());
                if (server != null) {
                    //預(yù)啟動(dòng)請(qǐng)求處理線(xiàn)程池，ThreadPoolExcutor.prestartAllCoreThreads()
                    server.getRequestProcessor().getRequestProcessThreadPool().prestartAllCoreThreads();
                    return server.getServerConfig();
                }
                return null;
            }
     }

跟進(jìn)AbstractServer.start

    public void start(ServerConfig serverConfig) {
                //請(qǐng)求處理線(xiàn)程池RequestThreadPoolProcessor
        requestProcessor = RequestProcessorFactory.selectProcessor();
                //調(diào)用NettyServer.doStart
        doStart(serverConfig);
        if (requestProcessor != null) {
                        //根據(jù)上篇ServerConfig的線(xiàn)程池參數(shù)配置創(chuàng)建DynamicThreadPool線(xiàn)程池
                        //底層實(shí)現(xiàn)為new ThreadPoolExecutor(...)
            requestProcessor.start(serverConfig);
        }
        this.serverConfig = serverConfig;
    }

接下來(lái)我們看下NettyServer實(shí)現(xiàn)。
該pigeon版本基于Netty3.9.2.Final，本篇不對(duì)netty服務(wù)做詳細(xì)說(shuō)明，感興趣的朋友可以閱讀下netty系列文章。

    public NettyServer() {
        this.bootstrap = new ServerBootstrap(channelFactory);
        this.bootstrap.setPipelineFactory(new NettyServerPipelineFactory(this));
        this.bootstrap.setOption("child.tcpNoDelay", true);
        this.bootstrap.setOption("child.keepAlive", true);
        this.bootstrap.setOption("child.reuseAddress", true);
        this.bootstrap.setOption("child.connectTimeoutMillis", 1000);
    }

    @Override
    public void doStart(ServerConfig serverConfig) {
        if (!started) {
            if (serverConfig.isAutoSelectPort()) {
                int availablePort = getAvailablePort(serverConfig.getPort());
                this.port = availablePort;
            } 
            InetSocketAddress address = null;
            if (this.ip == null) {
                address = new InetSocketAddress(this.port);
            } else {
                address = new InetSocketAddress(this.ip, this.port);
            }
            channel = this.bootstrap.bind(address);
            serverConfig.setActualPort(this.port);
            this.started = true;
        }
    }
    //在NettyServerPipelineFactory中定義了一些列消息處理器
    public ChannelPipeline getPipeline() {
        ChannelPipeline pipeline = pipeline();
        pipeline.addLast("framePrepender", new FramePrepender());
        pipeline.addLast("frameDecoder", new FrameDecoder());
        pipeline.addLast("crc32Handler", new Crc32Handler(codecConfig));
        pipeline.addLast("compressHandler", new CompressHandler(codecConfig));
        pipeline.addLast("providerDecoder", new ProviderDecoder());
        pipeline.addLast("providerEncoder", new ProviderEncoder());
        //NettyServerHandler處理器中處理InvocationRequest
        pipeline.addLast("serverHandler", new NettyServerHandler(server));
        return pipeline;
    }

因?yàn)楸酒谟诮榻B消息處理，我們重點(diǎn)看最后一個(gè)處理器NettyServerHandler，跟蹤代碼如下：

    //NettyServerHandler.messageReceived
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent message) {
        CodecEvent codecEvent = (CodecEvent) (message.getMessage());
        
        //請(qǐng)求信息
        InvocationRequest request = (InvocationRequest) codecEvent.getInvocation();
        ProviderContext invocationContext = new DefaultProviderContext(request, new NettyServerChannel(ctx.getChannel()));
        try {
            //請(qǐng)求處理
            this.server.processRequest(request, invocationContext);
        } catch (Throwable e) {
            String msg = "process request failed:" + request;
            // 心跳消息只返回正常的, 異常不返回
            if (request.getCallType() == Constants.CALLTYPE_REPLY
                    && request.getMessageType() != Constants.MESSAGE_TYPE_HEART) {
                ctx.getChannel().write(ProviderUtils.createFailResponse(request, e));
            }
        }
    }

    @Override
    public Future<InvocationResponse> processRequest(InvocationRequest request, ProviderContext providerContext) {
                //最終調(diào)用RequestThreadPoolProcessor.doProcessRequest
        return requestProcessor.processRequest(request, providerContext);
    }

RequestThreadPoolProcessor.doProcessRequest就是我們所有關(guān)注的最終的業(yè)務(wù)處理邏輯，幾點(diǎn)說(shuō)明如下：
1、將業(yè)務(wù)處理邏輯封裝在Callable任務(wù)中；
2、將Callable任務(wù)提交到線(xiàn)程池執(zhí)行；
3、ProviderProcessHandlerFactory中將不同消息類(lèi)型對(duì)應(yīng)的Filter集合以ServiceInvocationHandler封裝成責(zé)任鏈模式的消息處理器鏈
代碼如下：

    public Future<InvocationResponse> doProcessRequest(final InvocationRequest request,
                                                       final ProviderContext providerContext) {
        requestContextMap.put(request, providerContext);

        doMonitorData(request, providerContext);
        
        Callable<InvocationResponse> requestExecutor = new Callable<InvocationResponse>() {

            @Override
            public InvocationResponse call() throws Exception {
                providerContext.getTimeline().add(new TimePoint(TimePhase.T));
                try {
                    //根據(jù)消息類(lèi)型獲取對(duì)應(yīng)的Filter責(zé)任鏈
                    ServiceInvocationHandler invocationHandler = ProviderProcessHandlerFactory
                            .selectInvocationHandler(providerContext.getRequest().getMessageType());
                    if (invocationHandler != null) {
                        providerContext.setThread(Thread.currentThread());
                        //執(zhí)行責(zé)任鏈
                        return invocationHandler.handle(providerContext);
                    }
                } catch (Throwable t) {
                    logger.error("Process request failed with invocation handler, you should never be here.", t);
                } finally {
                    requestContextMap.remove(request);
                }
                return null;
            }
        };
        final ThreadPool pool = selectThreadPool(request);

        try {
            checkRequest(pool, request);
            providerContext.getTimeline().add(new TimePoint(TimePhase.T));
            //提交線(xiàn)程池執(zhí)行
            return pool.submit(requestExecutor);
        } catch (RejectedExecutionException e) {
            requestContextMap.remove(request);
            throw new RejectedException(getProcessorStatistics(pool), e);
        }

    }

關(guān)于ProviderProcessHandlerFactory中不同消息類(lèi)型對(duì)應(yīng)的Filter集合如下：

    public static void init() {
                //業(yè)務(wù)消息
        registerBizProcessFilter(new TraceFilter());
        if (Constants.MONITOR_ENABLE) {
            registerBizProcessFilter(new MonitorProcessFilter());
        }
        registerBizProcessFilter(new WriteResponseProcessFilter());
        registerBizProcessFilter(new ContextTransferProcessFilter());
        registerBizProcessFilter(new ExceptionProcessFilter());
        registerBizProcessFilter(new SecurityFilter());
        registerBizProcessFilter(new GatewayProcessFilter());
        registerBizProcessFilter(new BusinessProcessFilter());
        bizInvocationHandler = createInvocationHandler(bizProcessFilters);
                //心跳消息
        registerHeartBeatProcessFilter(new WriteResponseProcessFilter());
        registerHeartBeatProcessFilter(new HeartbeatProcessFilter());
        heartBeatInvocationHandler = createInvocationHandler(heartBeatProcessFilters);
                
        registerHealthCheckProcessFilter(new WriteResponseProcessFilter());
        registerHealthCheckProcessFilter(new HealthCheckProcessFilter());
        healthCheckInvocationHandler = createInvocationHandler(healthCheckProcessFilters);

        registerScannerHeartBeatProcessFilter(new WriteResponseProcessFilter());
        registerScannerHeartBeatProcessFilter(new ScannerHeartBeatProcessFilter());
        scannerHeartBeatInvocationHandler = createInvocationHandler(scannerHeartBeatProcessFilters);
    }

以業(yè)務(wù)消息對(duì)應(yīng)的Filter集合為例子，來(lái)看下責(zé)任鏈的封裝以及Filter的執(zhí)行順序

        //Filter逆序封裝，順序執(zhí)行
        //在ServiceInvocationHandler.handle中執(zhí)行filter.invoke，在filter.invoke中執(zhí)行next.handle
        private static <K, V extends ServiceInvocationFilter> ServiceInvocationHandler createInvocationHandler(
            List<V> internalFilters) {
        ServiceInvocationHandler last = null;
        List<V> filterList = new ArrayList<V>();
        filterList.addAll(internalFilters);
        for (int i = filterList.size() - 1; i >= 0; i--) {
            final V filter = filterList.get(i);
            final ServiceInvocationHandler next = last;
            last = new ServiceInvocationHandler() {
                @SuppressWarnings("unchecked")
                @Override
                public InvocationResponse handle(InvocationContext invocationContext) throws Throwable {
                    return filter.invoke(next, invocationContext);
                }
            };
        }
        return last;
    }

關(guān)于TraceFilter、MonitorProcessFilter、ExceptionProcessFilter等這里限于篇幅不具體分析代碼，這些Filter負(fù)責(zé)的功能從Filter命名可以猜個(gè)七七八八。我們重點(diǎn)看看BusinessProcessFilter

    @Override
    public InvocationResponse invoke(ServiceInvocationHandler handler, ProviderContext invocationContext)
            throws Throwable {
        invocationContext.getTimeline().add(new TimePoint(TimePhase.U));
        InvocationRequest request = invocationContext.getRequest();
        if (request.getMessageType() == Constants.MESSAGE_TYPE_SERVICE) {
            。。。
            InvocationResponse response = null;
            ServiceMethod method = invocationContext.getServiceMethod();
            if (method == null) {
                //ServiceMethodFactory緩存了service bean的方法映射關(guān)系
                method = ServiceMethodFactory.getMethod(request);
            }
            。。。
            Object returnObj = null;
            try {
                //基于反射調(diào)用
                returnObj = method.invoke(request.getParameters());
            } finally {
                ProviderHelper.clearContext();
            }

            invocationContext.getTimeline().add(new TimePoint(TimePhase.M, System.currentTimeMillis()));
            if (request.getCallType() == Constants.CALLTYPE_REPLY) {
                response = ProviderUtils.createSuccessResponse(request, returnObj);
            }
            return response;
        }
        throw new BadRequestException("message type[" + request.getMessageType() + "] is not supported!");
    }

可以看到我們根據(jù)請(qǐng)求信息InvocationRequest從ServiceMethodFactory找到對(duì)應(yīng)的service bean方法，然后基于反射執(zhí)行。
而ServiceMethodFactory在上篇s服務(wù)初始化過(guò)程中緩存了每一個(gè)service bean的方法映射關(guān)系。

至此我們理清了TCP協(xié)議NettyServer的消息處理機(jī)制。
關(guān)于http協(xié)議的JettyHttpServer實(shí)現(xiàn)不再贅述了，感興趣的朋友可以自行了解。

轉(zhuǎn)載請(qǐng)備注原文鏈接。