MyCAT源碼分析(三)請(qǐng)求處理流程

本篇將按請(qǐng)求處理的流程來分析MyCAT NIO網(wǎng)絡(luò)通信框架政钟。
從請(qǐng)求的接收開始:
1、NIOAcceptor 接收請(qǐng)求

    private void accept() {
        SocketChannel channel = null;
        try {
            channel = serverChannel.accept();
            channel.configureBlocking(false);
            FrontendConnection c = factory.make(channel);
            c.setAccepted(true);
            c.setId(ID_GENERATOR.getId());
            NIOProcessor processor = (NIOProcessor) MycatServer.getInstance()
                    .nextProcessor();
            c.setProcessor(processor);
            
            NIOReactor reactor = reactorPool.getNextReactor();
            reactor.postRegister(c);

        } catch (Exception e) {
            LOGGER.warn(getName(), e);
            closeChannel(channel);
        }
    }

拆解:
FrontendConnection c = factory.make(channel);
調(diào)用ServerConnectionFactory.make封裝SocketChannel為ServerConnection, 設(shè)置數(shù)據(jù)包handler處理器拖刃,創(chuàng)建當(dāng)前連接的NonBlockingSession

NIOReactor reactor = reactorPool.getNextReactor();
reactor.postRegister(c);
將前端連接分發(fā)到某一個(gè)reactor線程處理

2抖苦、NIOReactor注冊(cè)SelectionKey.OP_READ事件到selector洋侨,監(jiān)聽并處理

        private void register(Selector selector) {
            AbstractConnection c = null;
            if (registerQueue.isEmpty()) {
                return;
            }
            while ((c = registerQueue.poll()) != null) {
                try {
                    ((NIOSocketWR) c.getSocketWR()).register(selector);
                    c.register();
                } catch (Exception e) {
                    c.close("register err" + e.toString());
                }
            }
        }

        public void run() {
            final Selector selector = this.selector;
            Set<SelectionKey> keys = null;
            for (;;) {
                    selector.select(500L);
                    register(selector);
                    keys = selector.selectedKeys();
                    for (SelectionKey key : keys) {
                        AbstractConnection con = null;
                        Object att = key.attachment();
                        if (att != null) {
                            con = (AbstractConnection) att;
                            if (key.isValid() && key.isReadable()) {
                                    con.asynRead();
                                                        }
                                                 }
                                          }
                          }
                }

拆解:
((NIOSocketWR) c.getSocketWR()).register(selector);
注冊(cè)SelectionKey.OP_READ事件到selector

c.register();
同步發(fā)送握手?jǐn)?shù)據(jù)包矾瑰,并交給FrontendAuthenticator做前端認(rèn)證處理;認(rèn)證成功則設(shè)置下一個(gè)handler處理器FrontendCommandHandler酵镜,每一個(gè)前端handler都持有前端連接FrontendConnection

con.asynRead();
接收數(shù)據(jù)包碉碉,并交給FrontendCommandHandler處理

3、FrontendCommandHandler

    @Override
    public void handle(byte[] data)
    {
        if(source.getLoadDataInfileHandler()!=null&&source.getLoadDataInfileHandler().isStartLoadData())
        {
            MySQLMessage mm = new MySQLMessage(data);
            int  packetLength = mm.readUB3();
            if(packetLength+4==data.length)
            {
                source.loadDataInfileData(data);
            }
            return;
        }
        switch (data[4])
        {
            case MySQLPacket.COM_INIT_DB:
                commands.doInitDB();
                source.initDB(data);
                break;
            case MySQLPacket.COM_QUERY:
                commands.doQuery();
                source.query(data);
                break;
                淮韭。垢粮。。
                靠粪。蜡吧。毫蚓。
         }
  }

拆解:
if(source.getLoadDataInfileHandler()!=null&&source.getLoadDataInfileHandler().isStartLoadData())
or
switch (data[4])

根據(jù)不同的請(qǐng)求類型進(jìn)入不同的處理分支;
以MySQLPacket.COM_QUERY為例調(diào)用FrontendQueryHandler處理

4昔善、ServerQueryHandler處理查詢請(qǐng)求

    public void query(String sql) {
        
        ServerConnection c = this.source;
        int rs = ServerParse.parse(sql);
        int sqlType = rs & 0xff;
        
        switch (sqlType) {
        //explain sql
        case ServerParse.EXPLAIN:
            ExplainHandler.handle(sql, c, rs >>> 8);
            break;
        //explain2 datanode=? sql=?
        case ServerParse.EXPLAIN2:
            Explain2Handler.handle(sql, c, rs >>> 8);
            break;
        case ServerParse.SET:
            SetHandler.handle(sql, c, rs >>> 8);
            break;
        case ServerParse.SHOW:
            ShowHandler.handle(sql, c, rs >>> 8);
            break;
        case ServerParse.SELECT:
            SelectHandler.handle(sql, c, rs >>> 8);
            break;
                元潘。。君仆。
                翩概。。返咱。
         }

拆解:
根據(jù)不同的sql查詢類型钥庇,調(diào)用不同的handler;
以ServerParse.SELECT為例咖摹,調(diào)用SelectHandler评姨,并傳入ServerConnection

    public static void handle(String stmt, ServerConnection c, int offs) {
        int offset = offs;
        switch (ServerParseSelect.parse(stmt, offs)) {
        case ServerParseSelect.VERSION_COMMENT:
            SelectVersionComment.response(c);
            break;
        case ServerParseSelect.DATABASE:
            SelectDatabase.response(c);
            break;
        case ServerParseSelect.USER:
            SelectUser.response(c);
            break;
        default:
            c.execute(stmt, ServerParse.SELECT);
        }
        }

拆解:
查詢請(qǐng)求進(jìn)入分支c.execute(stmt, ServerParse.SELECT);
最終調(diào)用ServerConnection. routeEndExecuteSQL;

5、ServerConnection. routeEndExecuteSQL;

    public void routeEndExecuteSQL(String sql, int type, SchemaConfig schema) {
        // 路由計(jì)算
        RouteResultset rrs = null;
        try {
            rrs = MycatServer
                    .getInstance()
                    .getRouterservice()
                    .route(MycatServer.getInstance().getConfig().getSystem(),
                            schema, type, sql, this.charset, this);

        } catch (Exception e) {
            StringBuilder s = new StringBuilder();
            String msg = e.getMessage();
            writeErrMessage(ErrorCode.ER_PARSE_ERROR, msg == null ? e.getClass().getSimpleName() : msg);
            return;
        }
        if (rrs != null) {
            // session執(zhí)行
            session.execute(rrs, type);
        }
    }

拆解:
獲取路由節(jié)點(diǎn)信息萤晴,進(jìn)入session執(zhí)行

6参咙、NonBlockingSession.execute

    public void execute(RouteResultset rrs, int type) {
        RouteResultsetNode[] nodes = rrs.getNodes();
        if (nodes.length == 1) {
            singleNodeHandler = new SingleNodeHandler(rrs, this);
            singleNodeHandler.execute();
        } else {
            multiNodeHandler = new MultiNodeQueryHandler(type, rrs, autocommit, this);
            multiNodeHandler.execute();
        }
    }

拆解:
根據(jù)路由信息分別進(jìn)入單分片節(jié)點(diǎn)處理器和多分片節(jié)點(diǎn)處理器
以SingleNodeHandler為例向下分析

7、SingleNodeHandler.execute

    public void execute() throws Exception {
        startTime=System.currentTimeMillis();
        ServerConnection sc = session.getSource();
        this.isRunning = true;
        this.packetId = 0;
//      final BackendConnection conn = session.getTarget(node);
        BackendConnection tConn = session.getTarget(node);
        if (session.getSource().isLocked()) {
            if (tConn == null) {
                tConn = session.getLockedTarget(node); 
            }
        }
        final BackendConnection conn = tConn;
        node.setRunOnSlave(rrs.getRunOnSlave());    // 實(shí)現(xiàn) master/slave注解        
        _execute(conn);
        }

    private void _execute(BackendConnection conn) {
        if (session.closed()) {
            endRunning();
            session.clearResources(true);
            return;
        }
        conn.setResponseHandler(this);
        try {
            conn.execute(node, session.getSource(), session.getSource()
                    .isAutocommit());
        } catch (Exception e1) {
            executeException(conn, e1);
            return;
        }
    }

拆解:
conn.setResponseHandler(this);
設(shè)置MysqlConnection的SelectionKey.OP_READ事件處理器

conn.execute(node, session.getSource(), session.getSource()
.isAutocommit());
調(diào)用MysqlConnection.synAndDoExecute同步發(fā)送查詢數(shù)據(jù)包

8硫眯、從MysqlConnection.setResponseHandler(this);逆向分析對(duì)分片節(jié)點(diǎn)返回?cái)?shù)據(jù)包的處理;
核心代碼如下:

    public boolean setResponseHandler(ResponseHandler queryHandler) {
            ((MySQLConnectionHandler) handler).setResponseHandler(queryHandler);
            respHandler = queryHandler;
        }

MySQLConnectionAuthenticator.handle;
分片節(jié)點(diǎn)認(rèn)證成功則設(shè)置handler為MySQLConnectionHandler

    @Override
    public void handle(byte[] data) {
            switch (data[4]) {
            case OkPacket.FIELD_COUNT:
                HandshakePacket packet = source.getHandshake();
                if (packet == null) {
                    processHandShakePacket(data);
                    // 發(fā)送認(rèn)證數(shù)據(jù)包
                    source.authenticate();
                    break;
                }
                // 處理認(rèn)證結(jié)果
                source.setHandler(new MySQLConnectionHandler(source));
                source.setAuthenticated(true);
                                择同。两入。。
                                敲才。裹纳。。
                        }
        }

MySQLConnectionFactory.make

    public MySQLConnection make(MySQLDataSource pool, ResponseHandler handler,
            String schema) throws IOException {

        DBHostConfig dsc = pool.getConfig();
        NetworkChannel channel = openSocketChannel(MycatServer.getInstance()
                .isAIO());

        MySQLConnection c = new MySQLConnection(channel, pool.isReadNode());
        MycatServer.getInstance().getConfig().setSocketParams(c, false);
        c.setHost(dsc.getIp());
        c.setPort(dsc.getPort());
        c.setUser(dsc.getUser());
        c.setPassword(dsc.getPassword());
        c.setSchema(schema);
        c.setHandler(new MySQLConnectionAuthenticator(c, handler));
        c.setPool(pool);
        c.setIdleTimeout(pool.getConfig().getIdleTimeout());
        if (channel instanceof AsynchronousSocketChannel) {
            ((AsynchronousSocketChannel) channel).connect(
                    new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
                    (CompletionHandler) MycatServer.getInstance()
                            .getConnector());
        } else {
            ((NIOConnector) MycatServer.getInstance().getConnector())
                    .postConnect(c);

        }
        return c;
    }

拆解:
c.setHandler(new MySQLConnectionAuthenticator(c, handler));
設(shè)置MySQLConnectionAuthenticator為認(rèn)證處理器紧武;

((NIOConnector) MycatServer.getInstance().getConnector())
.postConnect(c);
將MySQLConnection注冊(cè)到NIOConnector剃氧,并設(shè)置attr為c,如下:
channel.register(selector, SelectionKey.OP_CONNECT, c);

收到分片節(jié)點(diǎn)的查詢結(jié)果的回調(diào)順序?yàn)椋?/p>

MySQLConnectionHandler.handle --> ResponseHandler.okResponse || errorResponse
--> SingleNodeHandler.session.getSource(); OkPacket.write(source)

直到這里才算理清了請(qǐng)求的處理流程阻星,以及前后端連接的關(guān)聯(lián)模式朋鞍。

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

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
禁止轉(zhuǎn)載妥箕,如需轉(zhuǎn)載請(qǐng)通過簡信或評(píng)論聯(lián)系作者滥酥。
  • 序言:七十年代末,一起剝皮案震驚了整個(gè)濱河市畦幢,隨后出現(xiàn)的幾起案子坎吻,更是在濱河造成了極大的恐慌,老刑警劉巖宇葱,帶你破解...
    沈念sama閱讀 222,681評(píng)論 6 517
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件瘦真,死亡現(xiàn)場(chǎng)離奇詭異刊头,居然都是意外死亡,警方通過查閱死者的電腦和手機(jī)诸尽,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,205評(píng)論 3 399
  • 文/潘曉璐 我一進(jìn)店門原杂,熙熙樓的掌柜王于貴愁眉苦臉地迎上來,“玉大人弦讽,你說我怎么就攤上這事污尉。” “怎么了往产?”我有些...
    開封第一講書人閱讀 169,421評(píng)論 0 362
  • 文/不壞的土叔 我叫張陵被碗,是天一觀的道長。 經(jīng)常有香客問我仿村,道長锐朴,這世上最難降的妖魔是什么? 我笑而不...
    開封第一講書人閱讀 60,114評(píng)論 1 300
  • 正文 為了忘掉前任蔼囊,我火速辦了婚禮焚志,結(jié)果婚禮上,老公的妹妹穿的比我還像新娘畏鼓。我一直安慰自己酱酬,他們只是感情好,可當(dāng)我...
    茶點(diǎn)故事閱讀 69,116評(píng)論 6 398
  • 文/花漫 我一把揭開白布云矫。 她就那樣靜靜地躺著膳沽,像睡著了一般。 火紅的嫁衣襯著肌膚如雪让禀。 梳的紋絲不亂的頭發(fā)上挑社,一...
    開封第一講書人閱讀 52,713評(píng)論 1 312
  • 那天,我揣著相機(jī)與錄音巡揍,去河邊找鬼痛阻。 笑死,一個(gè)胖子當(dāng)著我的面吹牛腮敌,可吹牛的內(nèi)容都是我干的阱当。 我是一名探鬼主播,決...
    沈念sama閱讀 41,170評(píng)論 3 422
  • 文/蒼蘭香墨 我猛地睜開眼糜工,長吁一口氣:“原來是場(chǎng)噩夢(mèng)啊……” “哼斗这!你這毒婦竟也來了?” 一聲冷哼從身側(cè)響起啤斗,我...
    開封第一講書人閱讀 40,116評(píng)論 0 277
  • 序言:老撾萬榮一對(duì)情侶失蹤表箭,失蹤者是張志新(化名)和其女友劉穎,沒想到半個(gè)月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體免钻,經(jīng)...
    沈念sama閱讀 46,651評(píng)論 1 320
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡彼水,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 38,714評(píng)論 3 342
  • 正文 我和宋清朗相戀三年,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了极舔。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片凤覆。...
    茶點(diǎn)故事閱讀 40,865評(píng)論 1 353
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡,死狀恐怖拆魏,靈堂內(nèi)的尸體忽然破棺而出盯桦,到底是詐尸還是另有隱情,我是刑警寧澤渤刃,帶...
    沈念sama閱讀 36,527評(píng)論 5 351
  • 正文 年R本政府宣布拥峦,位于F島的核電站,受9級(jí)特大地震影響卖子,放射性物質(zhì)發(fā)生泄漏略号。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 42,211評(píng)論 3 336
  • 文/蒙蒙 一洋闽、第九天 我趴在偏房一處隱蔽的房頂上張望玄柠。 院中可真熱鬧,春花似錦诫舅、人聲如沸羽利。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,699評(píng)論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽这弧。三九已至,卻和暖如春俏讹,著一層夾襖步出監(jiān)牢的瞬間,已是汗流浹背畜吊。 一陣腳步聲響...
    開封第一講書人閱讀 33,814評(píng)論 1 274
  • 我被黑心中介騙來泰國打工泽疆, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留,地道東北人玲献。 一個(gè)月前我還...
    沈念sama閱讀 49,299評(píng)論 3 379
  • 正文 我出身青樓殉疼,卻偏偏與公主長得像,于是被迫代替她去往敵國和親捌年。 傳聞我的和親對(duì)象是個(gè)殘疾皇子瓢娜,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 45,870評(píng)論 2 361

推薦閱讀更多精彩內(nèi)容