最新博客
最新资源下载
数据库路由中间件MyCat - 源代码篇（3）
勿忘初心2018-11-02 09:25此文已由作者张镐薪授权网易云社区发布。
欢迎访问网易云社区，了解更多网易技术产品运营经验。
2. 前端连接建立与认证
Title:MySql连接建立以及认证过程client->MySql:1.TCP连接请求 MySql->client:2.接受TCP连接client->MySql:3.TCP连接建立MySql->client:4.握手包HandshakePacketclient->MySql:5.认证包AuthPacketMySql->client:6.如果验证成功，则返回OkPacketclient->MySql:7.默认会发送查询版本信息的包MySql->client:8.返回结果包
2.3 （5~6）认证包AuthPacket，如果验证成功，则返回OkPacket
继续执行FrontendConnection的register()方法：
// 异步读取并处理，这个与RW线程中的asynRead()相同，之后客户端收到握手包返回AuthPacket就是从这里开始看。 this.asynRead();
FrontendConnection.asynRead()方法直接调用this.socketWR.asynRead();如之前所述，一个Connection对应一个socketWR。在这里是一个FrontendConnection对应一个NIOSocketWR。NIOSocketWR是操作类，里面的方法实现异步读写。 NIOSocketWR.asynRead():
public void asynRead() throws IOException {
ByteBuffer theBuffer = con.readBuffer; if (theBuffer == null) {
theBuffer = con.processor.getBufferPool().allocate();
con.readBuffer = theBuffer;
} //从channel中读取数据，并且保存到对应AbstractConnection的readBuffer中，readBuffer处于write mode，返回读取了多少字节
int got = channel.read(theBuffer); //调用处理读取到的数据的方法
con.onReadData(got);
}
读取完数据到缓存readBuffer后，调用处理readBuffer方法： AbstractConnection.onReadData(int got)：
public void onReadData(int got) throws IOException { //如果连接已经关闭，则不处理
if (isClosed.get()) { return;
}
ByteBuffer buffer = this.readBuffer;
lastReadTime = TimeUtil.currentTimeMillis(); //读取到的字节小于0，表示流关闭，如果等于0，代表TCP连接关闭了
if (got < 0) { this.close("stream closed");
return;
} else if (got == 0) {
if (!this.channel.isOpen()) {
this.close("socket closed");
return;
}
}
netInBytes += got;
processor.addNetInBytes(got); // 循环处理字节信息
//readBuffer一直处于write mode，position记录最后的写入位置
int offset = readBufferOffset, length = 0, position = buffer.position();
for (; ; ) { //获取包头的包长度信息
length = getPacketLength(buffer, offset);
if (length == -1) {
if (!buffer.hasRemaining()) {
buffer = checkReadBuffer(buffer, offset, position);
} break;
} //如果postion小于包起始位置加上包长度，证明readBuffer不够大，需要扩容
if (position >= offset + length) {
buffer.position(offset);
byte[] data = new byte[length]; //读取一个完整的包
buffer.get(data, 0, length); //处理包，每种AbstractConnection的处理函数不同
handle(data); //记录下读取到哪里了
offset += length; //如果最后写入位置等于最后读取位置，则证明所有的处理完了，可以清空缓存和offset
//否则，记录下最新的offset
//由于readBufferOffset只会单线程（绑定的RW线程）修改，但是会有多个线程访问（定时线程池的清理任务），所以设为volatile，不用CAS
if (position == offset) {
if (readBufferOffset != 0) {
readBufferOffset = 0;
}
buffer.clear();
break;
} else {
readBufferOffset = offset;
buffer.position(position);
continue;
}
} else {
if (!buffer.hasRemaining()) {
buffer = checkReadBuffer(buffer, offset, position);
} break;
}
}
}private ByteBuffer checkReadBuffer(ByteBuffer buffer, int offset,
int position) {
if (offset == 0) {
if (buffer.capacity() >= maxPacketSize) {
throw new IllegalArgumentException(
"Packet size over the limit.");
} int size = buffer.capacity() << 1;
size = (size > maxPacketSize) ? maxPacketSize : size;
ByteBuffer newBuffer = processor.getBufferPool().allocate(size);
buffer.position(offset);
newBuffer.put(buffer);
readBuffer = newBuffer;
recycle(buffer);
return newBuffer;
} else {
buffer.position(offset);
buffer.compact();
readBufferOffset = 0;
return buffer;
}
}
可以看出，处理缓存需要考虑到容量，扩容和位置记录等方面。 这里，readBuffer一直处于写模式。MyCat通过position()，还有get(data, 0, length)来读取数据。readBufferOffset用来记录每次读取的offset，需要设置为volatile。由于readBufferOffset只会单线程（绑定的RW线程）修改，但是会有多个线程访问（定时线程池的清理空闲连接的任务），所以设为volatile，不用CAS。这是一个经典的用volatile代替CAS实现多线程安全访问的场景。 MyCat的缓存管理思路很好，之后我会仔细讲。 读取完整包之后，交给handler处理。每种AbstractConnection的handler不同，FrontendConnection的handler为FrontendAuthenticator
this.handler = new FrontendAuthenticator(this);
我们思考下，FrontendConnection会接收什么请求呢？有两种，认证请求和SQL命令请求。只有认证成功后，才会接受SQL命令请求。FrontendAuthenticator只负责认证请求，在认证成功后，将对应AbstractConnection的handler设为处理SQL请求的FrontendCommandHandler即可. 一切正常的话，这里读取到的应为客户端发出的AuthPacket： AuthPacket:
packet length(3 bytes)
packet number (1)
client flags (4)
max packet (4)
charset(1)
username (null terminated string)
password (length code Binary)
database (null terminated string)
public void handle(byte[] data) { // check quit packet
if (data.length == QuitPacket.QUIT.length && data[4] == MySQLPacket.COM_QUIT) {
source.close("quit packet");
return;
}
AuthPacket auth = new AuthPacket();
auth.read(data); // check user
if (!checkUser(auth.user, source.getHost())) {
failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "' with host '" + source.getHost()+ "'");
return;
} // check password
if (!checkPassword(auth.password, auth.user)) {
failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "', because password is error "); return;
} // check degrade
if ( isDegrade( auth.user ) ) {
failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "', because service be degraded "); return;
} // check schema
switch (checkSchema(auth.database, auth.user)) {
case ErrorCode.ER_BAD_DB_ERROR:
failure(ErrorCode.ER_BAD_DB_ERROR, "Unknown database '" + auth.database + "'");
break; case ErrorCode.ER_DBACCESS_DENIED_ERROR:
String s = "Access denied for user '" + auth.user + "' to database '" + auth.database + "'";
failure(ErrorCode.ER_DBACCESS_DENIED_ERROR, s);
break;
default: //认证成功，设置好用户数据库和权限等，将handler设置为FrontendCommandHandler
success(auth);
}
}
认证成功后，会发送OkPacket，对应FrontendConnection的handler变成了FrontendCommandHandler，可以接受SQL请求了。 发送OkPacket的过程与HandshakePacket相同，这里不再赘述。
source.write(source.writeToBuffer(AUTH_OK, buffer));
OkPacket结构：
packet length(3 bytes)
packet number (1)
0x00 (1，包体首字节为0)
affect rows (length code Binary)
insert id (length code Binary)
server status (2)
warning status (2)
message (length code Binary)
更多网易技术、产品、运营经验分享请点击。
相关文章：
【推荐】 使用 MongoDB 存储商品分类信息
【推荐】 NOS服务监控实践
【推荐】 Jmeter入门实例
