資料庫路由中介軟體MyCat - 原始碼篇（8）
阿新 • • 發佈：2018-11-02
此文已由作者張鎬薪授權網易雲社群釋出。
歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。
3. 連線模組
3.5 後端連線
對於後端連線，我們只關心MySQL的。 從後端連線工廠開始MySQLConnectionFactory.java：
public MySQLConnection make(MySQLDataSource pool, ResponseHandler handler,
String schema) throws IOException { //DBHost配置
DBHostConfig dsc = pool.getConfig(); //根據是否為NIO返回SocketChannel或者AIO的AsynchronousSocketChannel
NetworkChannel channel = openSocketChannel(MycatServer.getInstance()
.isAIO()); //新建MySQLConnection
MySQLConnection c = new MySQLConnection(channel, pool.isReadNode()); //根據配置初始化MySQLConnection
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); //目前實際連線還未建立，handler為MySQL連線認證MySQLConnectionAuthenticator
c.setHandler(new MySQLConnectionAuthenticator(c, handler));
c.setPool(pool);
c.setIdleTimeout(pool.getConfig().getIdleTimeout()); //AIO和NIO連線方式建立實際的MySQL連線
if (channel instanceof AsynchronousSocketChannel) {
((AsynchronousSocketChannel) channel).connect( new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
(CompletionHandler) MycatServer.getInstance()
.getConnector());
} else { //通過NIOConnector建立連線
((NIOConnector) MycatServer.getInstance().getConnector())
.postConnect(c);
} return c;
}通過NIOConnector建立實際連線的過程與前端連線的建立相似，也是先放在佇列中，之後由NIOConnector去建立連線。
public void postConnect(AbstractConnection c) {
connectQueue.offer(c);
selector.wakeup();
}public void run() { final Selector tSelector = this.selector; for (;;) {
++connectCount; try { //檢視有無連線就緒
tSelector.select(1000L);
connect(tSelector);
Set<SelectionKey> keys = tSelector.selectedKeys(); try { for (SelectionKey key : keys) {
Object att = key.attachment(); if (att != null && key.isValid() && key.isConnectable()) {
finishConnect(key, att);
} else {
key.cancel();
}
}
} finally {
keys.clear();
}
} catch (Exception e) {
LOGGER.warn(name, e);
}
}
}private void connect(Selector selector) {
AbstractConnection c = null; while ((c = connectQueue.poll()) != null) { try {
SocketChannel channel = (SocketChannel) c.getChannel(); //註冊OP_CONNECT監聽與後端連線是否真正建立
channel.register(selector, SelectionKey.OP_CONNECT, c); //主動連線
channel.connect(new InetSocketAddress(c.host, c.port));
} catch (Exception e) {
c.close(e.toString());
}
}
}private void finishConnect(SelectionKey key, Object att) {
BackendAIOConnection c = (BackendAIOConnection) att; try { if (finishConnect(c, (SocketChannel) c.channel)) { //做原生NIO連線是否完成的判斷和操作
clearSelectionKey(key);
c.setId(ID_GENERATOR.getId()); //繫結特定的NIOProcessor以作idle清理
NIOProcessor processor = MycatServer.getInstance()
.nextProcessor();
c.setProcessor(processor); //與特定NIOReactor繫結監聽讀寫
NIOReactor reactor = reactorPool.getNextReactor();
reactor.postRegister(c);
}
} catch (Exception e) { //如有異常，將key清空
clearSelectionKey(key);
c.close(e.toString());
c.onConnectFailed(e);
}
}private boolean finishConnect(AbstractConnection c, SocketChannel channel)
throws IOException { if (channel.isConnectionPending()) {
channel.finishConnect();
c.setLocalPort(channel.socket().getLocalPort()); return true;
} else { return false;
}
}private void clearSelectionKey(SelectionKey key) { if (key.isValid()) {
key.attach(null);
key.cancel();
}
}繫結到具體的NIOReactor之後，監聽讀事件。和之前講的前端連線建立過程類似。這次是後端MySQL主動發握手包。這時，讀事件就緒，NIOReactor中的RW執行緒會呼叫對應AbstractConnection（這裡是MySQLConnection）的handler的處理方法處理。這裡MySQLConnection中的handler參考工廠方法發現是MySQLConnectionAuthenticator。檢視handle方法：
/**
* MySQL 4.1版本之前是MySQL323加密,MySQL 4.1和之後的版本都是MySQLSHA1加密，在MySQL5.5的版本之後可以客戶端外掛式加密(這個MyCat實現)
* @see @http://dev.mysql.com/doc/internals/en/determining-authentication-method.html
*/@Overridepublic void handle(byte[] data) { try { switch (data[4]) { //如果是OkPacket，檢查是否認證成功
case OkPacket.FIELD_COUNT:
HandshakePacket packet = source.getHandshake(); if (packet == null) { //如果為null，證明連結第一次建立，處理
processHandShakePacket(data); // 傳送認證資料包
source.authenticate(); break;
} // 如果packet不為null，處理認證結果
//首先將連線設為已驗證並將handler改為MySQLConnectionHandler
source.setHandler(new MySQLConnectionHandler(source));
source.setAuthenticated(true); //判斷是否用了壓縮協議
boolean clientCompress = Capabilities.CLIENT_COMPRESS==(Capabilities.CLIENT_COMPRESS & packet.serverCapabilities); boolean usingCompress= MycatServer.getInstance().getConfig().getSystem().getUseCompression()==1 ; if(clientCompress&&usingCompress)
{
source.setSupportCompress(true);
} //設定ResponseHandler
if (listener != null) {
listener.connectionAcquired(source);
} break; //如果為ErrorPacket，則認證失敗
case ErrorPacket.FIELD_COUNT:
ErrorPacket err = new ErrorPacket();
err.read(data);
String errMsg = new String(err.message);
LOGGER.warn("can't connect to mysql server ,errmsg:"+errMsg+" "+source); //source.close(errMsg);
throw new ConnectionException(err.errno, errMsg); //如果是EOFPacket，則為MySQL 4.1版本，是MySQL323加密
case EOFPacket.FIELD_COUNT:
auth323(data[3]); break; default:
packet = source.getHandshake(); if (packet == null) {
processHandShakePacket(data); // 傳送認證資料包
source.authenticate(); break;
} else { throw new RuntimeException("Unknown Packet!");
}
}
} catch (RuntimeException e) { if (listener != null) {
listener.connectionError(e, source); return;
} throw e;
}
}在連線建立並認證後，MySQLConnectionHandler來處理這個連線的請求和相應。 MySQL服務端響應客戶端查詢請求的流程如下：可以分為三個階段： （第一階段）客戶端傳送查詢請求包COM_QUERY （command query packet），如果有結果集返回，且結果集不為空，則返回FieldCount（列數量）包；如果結果集為空，則返回OKPacket；如果命令有錯，則返回ERRPacket；如果是Load file data命令，則返回LOCAL_INFILE_Request。 （第二階段）如果有結果集返回，則先返回列集合，所有列返回完了之後，會返回EOFPacket；如果過程中出現錯誤，則返回錯誤包。 （第三階段）之後返回行記錄，返回全部行記錄之後，返回EOFPacket。如果有錯誤，回錯誤包。 MyCat實現原始碼如下：
protected void handleData(byte[] data) { switch (resultStatus) { //第一階段
case RESULT_STATUS_INIT:
switch (data[4]) { //返回OKPacket
case OkPacket.FIELD_COUNT:
handleOkPacket(data); break; //返回錯誤包
case ErrorPacket.FIELD_COUNT:
handleErrorPacket(data); break; //返回Load Data進一步操作
case RequestFilePacket.FIELD_COUNT:
handleRequestPacket(data); break; //返回結果集列數量 default:
//記錄列數量並進入第二階段
resultStatus = RESULT_STATUS_HEADER;
header = data;
fields = new ArrayList<byte[]>((int) ByteUtil.readLength(data, 4));
} break; //第二階段
case RESULT_STATUS_HEADER:
switch (data[4]) { //返回錯誤包
case ErrorPacket.FIELD_COUNT:
resultStatus = RESULT_STATUS_INIT;
handleErrorPacket(data); break; //返回EOF，證明列集合返回完畢，進入第三階段
case EOFPacket.FIELD_COUNT:
resultStatus = RESULT_STATUS_FIELD_EOF;
handleFieldEofPacket(data); break; //返回的是列集合，記錄 default:
fields.add(data);
} break; //第三階段
case RESULT_STATUS_FIELD_EOF:
switch (data[4]) { //返回錯誤包
case ErrorPacket.FIELD_COUNT:
resultStatus = RESULT_STATUS_INIT;
handleErrorPacket(data); break; //返回EOF，證明結果集返回完畢，回到第一階段等待下一個請求的響應
case EOFPacket.FIELD_COUNT:
resultStatus = RESULT_STATUS_INIT;
handleRowEofPacket(data); break; //返回結果集包 default:
handleRowPacket(data);
} break; default:
throw new RuntimeException("unknown status!");
}
}
更多網易技術、產品、運營經驗分享請點選。
相關文章：
【推薦】 雙11背後的黑科技：大資料實時計算如何為你量身定製？