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
*/
@Override
public 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!");
}
}
