2024-01-18  阅读(115)
原文作者:hashcon 原文地址: https://zhanghaoxin.blog.csdn.net/article/details/50682173

2. 前端连接建立与认证

Created with Raphaël 2.1.0MySql连接建立以及认证过程clientclientMySqlMySql1.TCP连接请求2.接受TCP连接3.TCP连接建立4.握手包HandshakePacket5.认证包AuthPacket6.如果验证成功,则返回OkPacket7.默认会发送查询版本信息的包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:

202401182020195871.png
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结构:

202401182020199742.png

  • 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)
阅读全文
  • 点赞