Cobar启动完成,监听特定端口。整个认证的流程图: NIOAcceptor类继承自Thread类,该类的对象会以线程的方式运行,进行连接的监听。NIOAcceptor启动的初始化过程如下:1 、打开一个selector,获取一个ServerSocketChannel对象,对该对象的socket绑定特定的监
Cobar启动完成,监听特定端口。整个认证的流程图: NIOAcceptor类继承自Thread类,该类的对象会以线程的方式运行,进行连接的监听。 NIOAcceptor启动的初始化过程如下: 1 、打开一个selector,获取一个ServerSocketChannel对象,对该对象的socket绑定特定的监听端口,并设置该channel为非阻塞模式,然后想selector注册该channel,绑定感兴趣的事件位OP_ACCEPT。public NIOAcceptor(String name, int port, FrontendConnectionFactory 分布式数据库中间件–(2) Cobar与客户端的握手认证) throws IOException { super.setName(name); this.port = port; this.selector = Selector.open(); this.serverChannel = ServerSocketChannel.open(); //ServerSocket使用TCP this.serverChannel.socket().bind(new InetSocketAddress(port)); this.serverChannel.configureBlocking(false); this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); this.分布式数据库中间件–(2) Cobar与客户端的握手认证 = 分布式数据库中间件–(2) Cobar与客户端的握手认证; }2、 然后会启动该线程,线程的run函数如下:
public void run() { final Selector selector = this.selector; //线程一直循环 for (;;) { ++acceptCount; try { selector.select(1000L); Set<SelectionKey> keys = selector.selectedKeys(); try { for (SelectionKey key : keys) { if (key.isValid() && key.isAcceptable()) { //接受来自客户端的连接 accept(); } else { key.cancel(); } } } finally { keys.clear(); } } catch (Throwable e) { LOGGER.warn(getName(), e); } } }3 、 该线程会一直循环监听想该selector注册过的server channel所感兴趣的事件(OP_ACCEPT),当有新的连接请求时,selector就会返回,keys就是请求连接的所有的包含channel的key集合。 SelectionKey有如下属性:
private void accept() { SocketChannel channel = null; try { //从服务器端获取管道,为一个新的连接返回channel channel = serverChannel.accept(); //配置管道为非阻塞 channel.configureBlocking(false); //前端连接工厂对管道进行配置,设置socket的收发缓冲区大小,TCP延迟等 //然后由成员变量分布式数据库中间件–(2) Cobar与客户端的握手认证的类型生产对于的类型的连接 //比如ServerConnectionFactory会返回ServerConnection实例,并对其属性进行设置 FrontendConnection c = 分布式数据库中间件–(2) Cobar与客户端的握手认证.make(channel); //设置连接属性 c.setAccepted(true); c.setId(ID_GENERATOR.getId()); //从processors中选择一个NIOProcessor,将其和该连接绑定 NIOProcessor processor = nextProcessor(); c.setProcessor(processor); //向读反应堆注册该连接,加入待处理队列 //select选择到感兴趣的事件后,会进行调用connection的read函数 processor.postRegister(c); } catch (Throwable e) { closeChannel(channel); LOGGER.warn(getName(), e); } }首先从serverchannel中accept后会返回一个socketchannel对象,然后设置该socket channel属性位非阻塞模式,然后将channel交给ServerConnectionFactory工厂,会产生一个ServerConnection对象。 FrontendConnectionFactory是一个抽象类,其中的getConnection方法是抽象方法,有具体子类连接工厂来实现。FrontendConnectionFactory的make方法对channel中的socket进行属性设置(接收和发送的缓冲区大小、延时、KeepAlive等),然后调用具体调用具体子类(ServerConnectionFactory)的getConnection来返回一个ServerConnection,返回后会在进行设置一下该ServerConnection的包头大小、最大包大小、设置连接的发送缓冲区队列、超时时间、字符编码,到此,工厂完成了新建连接的工作,返回一个连接的对象。返回后将该连接分配给一个processor,该processor会将该连接保存,processor也会对连接进行定期检查。 5、 processor还会向自己的reactorR进行注册该连接,加入reactorR的处理队列,并唤醒阻塞的select()方法。 反应堆中Reactor的R线程运行代码:
public void run() { final Selector selector = this.selector; for (;;) { ++reactCount; try { int res = selector.select(); LOGGER.debug(reactCount + ">>NIOReactor接受连接数:" + res); register(selector); Set<SelectionKey> keys = selector.selectedKeys(); try { for (SelectionKey key : keys) { Object att = key.attachment(); if (att != null && key.isValid()) { int readyOps = key.readyOps(); if ((readyOps & SelectionKey.OP_READ) != 0) { LOGGER.debug("select读事件"); read((NIOConnection) att); } else if ((readyOps & SelectionKey.OP_WRITE) != 0) { LOGGER.debug("select写事件"); write((NIOConnection) att); } else { key.cancel(); } } else { key.cancel(); } } } finally { keys.clear(); } } catch (Throwable e) { LOGGER.warn(name, e); } } }该R线程也会一直循环运行,如果向该selector注册过的channel没有对应的感兴趣的事件发生,就会阻塞,直到有感兴趣的事件发生或被wakeup。返回后会运行register函数,将之前加入该reactor连接队列中的所有连接向该selector注册OP_READ事件。该注册的动作会调用Connection对象中的register方法进行注册,并像客户端主动发送握手数据包(HandshakePacket)。
?channel.register(selector, SelectionKey.OP_READ, this);注意最后一个this指针参数,表示将该连接作为附件,注册到selector,当有感兴趣的时间发生时,函数selector.selectedKeys()返回的SelectionKey集合中的对象中使用key.attachment()即可获取到上面注册时绑定的connection对象指针附件。目的就是为了通过该附件对象调用该连接类中定义的read函数来完成功能。如下所示:
private void read(NIOConnection c) { try { c.read(); } catch (Throwable e) { c.error(ErrorCode.ERR_READ, e); } }6、 连接类中定义的read函数定义在AbstractConnection类中。在该read函数
public void handle(final byte[] data) { // 从线程池获取一个线程,异步处理前端数据 // 从processor中的线程池中获取一个可以执行的线程,执行Runnable任务 processor.getHandler().execute(new Runnable() { @Override public void run() { try { //调用具体NIOHandler子类的handle函数 handler.handle(data); } catch (Throwable t) { error(ErrorCode.ERR_HANDLE_DATA, t); } } }); }8、 handler在构造函数中初始化成前端认证处理器,用于处理前端权限认证。
public FrontendConnection(SocketChannel channel) { super(channel); ..................... //前端认证处理器 this.handler = new FrontendAuthenticator(this); }9、 由于Cobar是基于MySQL协议的,所以需要分析一下MySQL协议的具体格式。下面就先分析一下MySQL认证数据包的格式: 每个报文都分为消息头和消息体两部分,其中消息头是固定的四个字节,报文结构如下: 登录认证报文的报文数据部分格式如下: 10、 FrontendAuthenticator类对上面的数据包的具体处理如下:
public void handle(byte[] data) { // check quit packet if (data.length == QuitPacket.QUIT.length && data[4] == MySQLPacket.COM_QUIT) { source.close(); 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 + "'"); return; } // check password if (!checkPassword(auth.password, auth.user)) { failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "'"); 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); } }在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。认证成功后会执行:
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)); ....... ByteBuffer buffer = source.allocate(); source.write(source.writeToBuffer(AUTH_OK, buffer)); }可以看到,在上面的函数中,设置连接对象source中的成员(是否认证、用户、数据库、编码、处理该连接后续数据包的处理器【handle方法】) 然后回复认证成功的消息。后面客户端再发送消息,会交给前端命令处理器进行处理。 客户端进行链接的时候Cobar服务器的输出:
16:59:19,388 INFO =============================================== 16:59:19,389 INFO Cobar is ready to startup ... 16:59:19,389 INFO Startup processors ... 16:59:19,455 INFO Startup connector ... 16:59:19,460 INFO Initialize dataNodes ... 16:59:19,506 INFO dnTest1:0 init success 16:59:19,514 INFO dnTest3:0 init success 16:59:19,517 INFO dnTest2:0 init success 16:59:19,527 INFO CobarServer is started and listening on 8066 16:59:19,527 INFO =============================================== 16:59:23,459 DEBUG 1>>NIOReactor接受连接数:0 16:59:23,464 DEBUG 2>>NIOReactor接受连接数:1 16:59:23,465 DEBUG select读事件 16:59:23,465 INFO com.alibaba.cobar.net.handler.FrontendAuthenticator接收的请求长度:62 58 0 0 1 5 166 15 0 0 0 0 1 33 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 114 111 111 116 0 20 169 171 247 102 133 96 158 224 121 22 226 229 88 244 119 238 185 61 124 219 16:59:23,468 INFO [thread=Processor1-H0,class=ServerConnection,host=192.168.137.8,port=46101,schema=null]'root' login success客户端得到的回复:
yan@yan-Z400:~$ mysql -uroot -p** -P8066 -h192.168.137.8 Welcome to the MySQL monitor. Commands end with ; or \g. Your MySQL connection id is 1 Server version: 5.1.48-cobar-1.2.7 Cobar Server (ALIBABA) Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved. Oracle is a registered trademark of Oracle Corporation and/or its affiliates. Other names may be trademarks of their respective owners. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. mysql>MySQL客户端的命令处理,具体后续会分析。
本文出自:http://blog.geekcome.com, 原文地址:http://blog.geekcome.com/%e5%88%86%e5%b8%83%e5%bc%8f%e6%95%b0%e6%8d%ae%e5%ba%93%e4%b8%ad%e9%97%b4%e4%bb%b6-2-cobar%e4%b8%8e%e5%ae%a2%e6%88%b7%e7%ab%af%e7%9a%84%e6%8f%a1%e6%89%8b%e8%ae%a4%e8%af%81, 感谢原作者分享。