MyCat - 源代码篇（2）

数据库路由中间件MyCat - 源代码篇（2）

2. 前端连接建立与认证

Created with Raphaël 2.1.0 MySql连接建立以及认证过程 client client MySql MySql 1.TCP连接请求 2.接受TCP连接 3.TCP连接建立 4.握手包HandshakePacket 5.认证包AuthPacket 6.如果验证成功，则返回OkPacket 7.默认会发送查询版本信息的包 8.返回结果包

2.2 (4)握手包HandshakePacket

NIOReactor其实就是一个网络事件反应转发器。



很多地方会用到NIOReactor，这里先讲FrontendConnection和NIOReactor绑定这一部分。上一节说到，NIOAcceptor的accept()最后将FrontendConnection交给了NIOReactor池其中的一个NIOReactor。调用的是 postRegister(AbstractConnection c)方法。

final void postRegister(AbstractConnection c) {
reactorR.registerQueue.offer(c);
reactorR.selector.wakeup();
}

postRegister将刚才传入的FrontendConnection放入RW线程的注册队列。之后，唤醒RW线程的selector。
为什么放入RW线程的注册队列，而不是直接注册呢？如果是直接注册，那么就是NIOAcceptor这个线程负责注册，这里就会有锁竞争，因为NIOAcceptor这个线程和每个RW线程会去竞争selector的锁。这样NIOAcceptor就不能高效的处理连接。所以，更好的方式是将FrontendConnection放入RW线程的注册队列，之后让RW线程自己完成注册工作。
RW线程的源代码：

private final class RW implements Runnable {
private final Selector selector;
private final ConcurrentLinkedQueue<AbstractConnection> registerQueue;
private long reactCount;

private RW() throws IOException {
this.selector = Selector.open();
this.registerQueue = new ConcurrentLinkedQueue<AbstractConnection>();
}

@Override
public void run() {
final Selector selector = this.selector;
Set<SelectionKey> keys = null;
for (;;) {
++reactCount;
try {
selector.select(500L);
//从注册队列中取出AbstractConnection之后注册读事件
//之后做一些列操作，请参考下面注释
register(selector);
keys = selector.selectedKeys();
for (SelectionKey key : keys) {
AbstractConnection con = null;
try {
Object att = key.attachment();
if (att != null) {
con = (AbstractConnection) att;
if (key.isValid() && key.isReadable()) {
try {
//异步读取数据并处理数据
con.asynRead();
} catch (IOException e) {
con.close("program err:" + e.toString());
continue;
} catch (Exception e) {
LOGGER.debug("caught err:", e);
con.close("program err:" + e.toString());
continue;
}
}
if (key.isValid() && key.isWritable()) {
//异步写数据
con.doNextWriteCheck();
}
} else {
key.cancel();
}
} catch (CancelledKeyException e) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(con + " socket key canceled");
}
} catch (Exception e) {
LOGGER.warn(con + " " + e);
}
}
} catch (Exception e) {
LOGGER.warn(name, e);
} finally {
if (keys != null) {
keys.clear();
}

}
}
}

private void register(Selector selector) {
AbstractConnection c = null;
if (registerQueue.isEmpty()) {
return;
}
while ((c = registerQueue.poll()) != null) {
try {
//注册读事件
((NIOSocketWR) c.getSocketWR()).register(selector);
//连接注册，对于FrontendConnection是发送HandshakePacket并异步读取响应
//响应为AuthPacket，读取其中的信息，验证用户名密码等信息，如果符合条件
//则发送OkPacket
c.register();
} catch (Exception e) {
c.close("register err" + e.toString());
}
}
}

}

因为NIOAcceptor线程和RW线程这两个都会操作RW线程的注册队列，所以要用ConcurrentLinkedQueue
RW线程不断检查selector中需要响应的事件，并如果注册队列不为空，就不断注册其中的AbstractConnection，在这里就是FrontendConnection。
之后执行FrontendConnection的register()方法：

@Override
public void register() throws IOException {
if (!isClosed.get()) {

// 生成认证数据
byte[] rand1 = RandomUtil.randomBytes(8);
byte[] rand2 = RandomUtil.randomBytes(12);

// 保存认证数据
byte[] seed = new byte[rand1.length + rand2.length];
System.arraycopy(rand1, 0, seed, 0, rand1.length);
System.arraycopy(rand2, 0, seed, rand1.length, rand2.length);
this.seed = seed;

// 发送握手数据包
HandshakePacket hs = new HandshakePacket();
hs.packetId = 0;
hs.protocolVersion = Versions.PROTOCOL_VERSION;
hs.serverVersion = Versions.SERVER_VERSION;
hs.threadId = id;
hs.seed = rand1;
hs.serverCapabilities = getServerCapabilities();
hs.serverCharsetIndex = (byte) (charsetIndex & 0xff);
hs.serverStatus = 2;
hs.restOfScrambleBuff = rand2;
// 异步写，本节就讲到这里
hs.write(this);

// 异步读取并处理，这个与RW线程中的asynRead()相同，之后客户端收到握手包返回AuthPacket（就是下一节）就是从这里开始看。
this.asynRead();
}
}

这个方法就是生成HandshakePacket并发送出去，之后异步读取响应。
之前的示例中MySql的HandshakePacket结构：



可以总结出：
HandshakePacket：
- packet length(3 bytes)
- packet number (1)
- protocol version (1)
- version (null terminated string)
- thread id (4)
- salt (8)
- server capabilities (2)
- server charset (1)
- server status (2)
- unused (13)
- salt (12)
- 0x00 — 结束

这里我们看下MyCat中的实现这一部分MySql协议栈的packet类结构：



这里可以看出，每个包都实现了自己的包长度和信息方法，并且针对前段后端连接都有读写方法实现，所以，之后读写数据都会根据场景不同调用这些类中的方法。这些包就是整个MySql协议栈除逻辑外的内容实现。
HandshakePacket.write(FrontendConnection c)方法将上面传入的数据封装成ByteBuffer，并传入给FrontendConnection c的write（ByteBuffer buffer），这个方法直接继承自AbstractConnection：

public final void write(ByteBuffer buffer) {
//首先判断是否为压缩协议
if(isSupportCompress())
{
//CompressUtil为压缩协议辅助工具类
ByteBuffer     newBuffer= CompressUtil.compressMysqlPacket(buffer,this,compressUnfinishedDataQueue);
//将要写的数据先放入写缓存队列
writeQueue.offer(newBuffer);

}   else
{
//将要写的数据先放入写缓存队列
writeQueue.offer(buffer);
}
try {
//处理写事件，这个方法比较复杂，需要重点分析其思路
this.socketWR.doNextWriteCheck();
} catch (Exception e) {
LOGGER.warn("write err:", e);
this.close("write err:" + e);

}

}

如代码注释中所述，先将要写的数据放入写缓冲队列，之后调用NIOSocketWR.doNextWriteCheck()处理写事件。

public void doNextWriteCheck() {
//检查是否正在写,看CAS更新writing值是否成功
if (!writing.compareAndSet(false, true)) {
return;
}

try {
//利用缓存队列和写缓冲记录保证写的可靠性，返回true则为全部写入成功
boolean noMoreData = write0();
//因为只有一个线程可以成功CAS更新writing值，所以这里不用再CAS
writing.set(false);
//如果全部写入成功而且写入队列为空（有可能在写入过程中又有新的Bytebuffer加入到队列），则取消注册写事件
//否则，继续注册写事件
if (noMoreData && con.writeQueue.isEmpty()) {
if ((processKey.isValid() && (processKey.interestOps() & SelectionKey.OP_WRITE) != 0)) {
disableWrite();
}

} else {

if ((processKey.isValid() && (processKey.interestOps() & SelectionKey.OP_WRITE) == 0)) {
enableWrite(false);
}
}

} catch (IOException e) {
if (AbstractConnection.LOGGER.isDebugEnabled()) {
AbstractConnection.LOGGER.debug("caught err:", e);
}
con.close("err:" + e);
}

}

private boolean write0() throws IOException {
int written = 0;
ByteBuffer buffer = con.writeBuffer;
if (buffer != null) {
//只要写缓冲记录中还有数据就不停写入，但如果写入字节为0，证明网络繁忙，则退出
while (buffer.hasRemaining()) {
written = channel.write(buffer);
if (written > 0) {
con.netOutBytes += written;
con.processor.addNetOutBytes(written);
con.lastWriteTime = TimeUtil.currentTimeMillis();
} else {
break;
}
}
//如果写缓冲中还有数据证明网络繁忙，计数并退出，否则清空缓冲
if (buffer.hasRemaining()) {
con.writeAttempts++;
return false;
} else {
con.writeBuffer = null;
con.recycle(buffer);
}
}
//读取缓存队列并写channel
while ((buffer = con.writeQueue.poll()) != null) {
if (buffer.limit() == 0) {
con.recycle(buffer);
con.close("quit send");
return true;
}
buffer.flip();
while (buffer.hasRemaining()) {
written = channel.write(buffer);
if (written > 0) {
con.lastWriteTime = TimeUtil.currentTimeMillis();
con.netOutBytes += written;
con.processor.addNetOutBytes(written);
con.lastWriteTime = TimeUtil.currentTimeMillis();
} else {
break;
}
}
//如果写缓冲中还有数据证明网络繁忙，计数，记录下这次未写完的数据到写缓冲记录并退出，否则回收缓冲
if (buffer.hasRemaining()) {
con.writeBuffer = buffer;
con.writeAttempts++;
return false;
} else {
con.recycle(buffer);
}
}
return true;
}

private void disableWrite() {
try {
SelectionKey key = this.processKey;
key.interestOps(key.interestOps() & OP_NOT_WRITE);
} catch (Exception e) {
AbstractConnection.LOGGER.warn("can't disable write " + e + " con "
+ con);
}

}

private void enableWrite(boolean wakeup) {
boolean needWakeup = false;
try {
SelectionKey key = this.processKey;
key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
needWakeup = true;
} catch (Exception e) {
AbstractConnection.LOGGER.warn("can't enable write " + e);

}
if (needWakeup && wakeup) {
processKey.selector().wakeup();
}
}

注释已经很详细，如此执行完，便成功将握手包发送给了客户端。
在这里稍微吐槽下，由于MyCat在网络通信上同时做了AIO和NIO，但是在设计上AbstractionConnection和这些并没有关系。但是又涉及到缓存队列，所以设计上出现了一些如下的类模式：



这样应该是不推荐这么设计的，目前我还没想好如何去改善