更多MyCat源码分析，请戳MyCat源码分析系列
MyCat前端验证
MyCat的前端验证指的是应用连接MyCat时进行的用户验证过程，如使用MySQL客户端时，$ mysql -uroot -proot -P8066 db_test触发的一系列行为。
验证的过程分为几个步骤：
1）应用与MyCat建立TCP连接；
2）MyCat发送握手包，其中带有为密码加密的种子（seed）；
3）应用接收握手包，使用种子进行密码加密，随后将包含用户名、加密密码、字符集和需连接的数据库（可选）等的验证包发送给MyCat；
4）MyCat依次验证信息，并将验证结果回发给应用
MyCat中，前端应用与MyCat之间建立的连接称为FrontendConnection（其子类为ServerConnection和ManagerConnection，分别由ServerConnectionFactory和ManagerConnectionFactory负责创建），在构造函数的时候绑定了一个NIOHandler类型的FrontendAuthenticator用于后续的验证
public FrontendConnection(NetworkChannel channel) throws IOException {
super(channel);
InetSocketAddress localAddr = (InetSocketAddress) channel.getLocalAddress();
InetSocketAddress remoteAddr = null;
if (channel instanceof SocketChannel) {
remoteAddr = (InetSocketAddress) ((SocketChannel) channel).getRemoteAddress();
} else if (channel instanceof AsynchronousSocketChannel) {
remoteAddr = (InetSocketAddress) ((AsynchronousSocketChannel) channel).getRemoteAddress();
}
this.host = remoteAddr.getHostString();
this.port = localAddr.getPort();
this.localPort = remoteAddr.getPort();
this.handler = new FrontendAuthenticator(this);
}
首先，应用与MyCat的TCP连接建立过程由NIOAcceptor负责完成，当之前注册的OP_ACCEPT事件得到响应时调用accept()方法：
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);
}
}
该TCP连接建立完成后创建一个FrontendConnection实例，并交由NIOReactor负责后续的读写工作（reactor.postRegister(c);），而NIOReactor中的内部类RW会调用其register()方法：
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());
}
}
}
其中，NIOSocketWR会调用其register()方法注册OP_READ事件，随后FrontendConnection调用其register()方法发送握手包HandshakePacket，等待应用回发验证包：
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);
// asynread response
this.asynRead();
}
}
当收到应用的验证包AuthPacket时，之前绑定在FrontendConnection上的FrontendAuthenticator会调用其handle()方法：
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:
success(auth);
}
}
依次验证用户名、密码、用户负载、数据库权限，验证成功后调用success()方法向应用发送OK包：
protected void success(AuthPacket auth) {
source.setAuthenticated(true);
source.setUser(auth.user);
source.setSchema(auth.database);
source.setCharsetIndex(auth.charsetIndex);
source.setHandler(new FrontendCommandHandler(source));
if (LOGGER.isInfoEnabled()) {
StringBuilder s = new StringBuilder();
s.append(source).append('\'').append(auth.user).append("' login success");
byte[] extra = auth.extra;
if (extra != null && extra.length > 0) {
s.append(",extra:").append(new String(extra));
}
LOGGER.info(s.toString());
}
ByteBuffer buffer = source.allocate();
source.write(source.writeToBuffer(AUTH_OK, buffer));
boolean clientCompress = Capabilities.CLIENT_COMPRESS==(Capabilities.CLIENT_COMPRESS & auth.clientFlags);
boolean usingCompress= MycatServer.getInstance().getConfig().getSystem().getUseCompression()==1 ;
if(clientCompress&&usingCompress)
{
source.setSupportCompress(true);
}
}
这里最重要的就是将FrontendConnection的handler重新绑定为FrontendCommandHandler对象，用于后续各类命令的接收和处理分发。
后端验证
后端验证过程类似于前端验证过程，区别就在于此时MyCat是作为客户端，向后端MySQL数据库发起连接与验证请求。
MyCat中，MyCat与后端MySQL的连接称为MySQLConnection，由MySQLConnectionFactory创建，并绑定了一个NIOHandler类型的MySQLConnectionAuthenticator，用于完成与MySQL的验证过程：
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;
}
MySQLConnection创建完成后，交由NIOConnector完成TCP连接，它会调用其connect()方法注册OP_CONNECT事件，并发送连接请求至MySQL：
private void connect(Selector selector) {
AbstractConnection c = null;
while ((c = connectQueue.poll()) != null) {
try {
SocketChannel channel = (SocketChannel) c.getChannel();
channel.register(selector, SelectionKey.OP_CONNECT, c);
channel.connect(new InetSocketAddress(c.host, c.port));
} catch (Exception e) {
c.close(e.toString());
}
}
}
TCP连接建立后同样交由NIOReactor负责后续的读写工作，最终之前绑定在MySQLConnection上的MySQLConnectionAuthenticator会调用其handle()方法：
public void handle(byte[] data) {
try {
switch (data[4]) {
case OkPacket.FIELD_COUNT:
HandshakePacket packet = source.getHandshake();
if (packet == null) {
processHandShakePacket(data);
// 发送认证数据包
source.authenticate();
break;
}
// 处理认证结果
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);
}
if (listener != null) {
listener.connectionAcquired(source);
}
break;
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);
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;
}
}
与前端验证一样，MySQL数据库作为服务端会在TCP连接建立完成后向应用发送一个握手包HandshakePacket，在此MySQLConnectionAuthenticator负责读取此包，并通过source.authenticate();调用MySQLConnection的authenticate()方法向MySQL发送验证包AuthPacket：
public void authenticate() {
AuthPacket packet = new AuthPacket();
packet.packetId = 1;
packet.clientFlags = clientFlags;
packet.maxPacketSize = maxPacketSize;
packet.charsetIndex = this.charsetIndex;
packet.user = user;
try {
packet.password = passwd(password, handshake);
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException(e.getMessage());
}
packet.database = schema;
packet.write(this);
}
验证通过后，将MySQLConnection的handler重新绑定为MySQLConnectionHandler对象，用于后续接收MySQL数据库发送过来的各类数据和处理分发
参考资料：
[1] http://sonymoon.iteye.com/blog/2245141
为尊重原创成果，如需转载烦请注明本文出处：http://www.cnblogs.com/fernandolee24/p/5196332.html，特此感谢