本篇將按請(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)備注原文鏈接。
