Cobar启动完成，监听特定端口。整个认证的流程图： NIOAcceptor类继承自Thread类，该类的对象会以线程的方式运行，进行连接的监听。 NIOAcceptor启动的初始化过程如下： 1 、打开一个selector，获取一个ServerSocketChannel对象，对该对象的socket绑定特定的
Cobar启动完成，监听特定端口。整个认证的流程图：
NIOAcceptor类继承自Thread类，该类的对象会以线程的方式运行，进行连接的监听。
NIOAcceptor启动的初始化过程如下：
1 、打开一个selector，获取一个ServerSocketChannel对象，对该对象的socket绑定特定的监听端口，并设置该channel为非阻塞模式，然后想selector注册该channel，绑定感兴趣的事件位OP_ACCEPT。
| 01 |
public <code class="Brush plain">NIOAcceptor(String name, <code class="Brush keyword">int <code class="Brush plain">port, FrontendConnectionFactory factory) <code class="Brush keyword">throws <code class="Brush plain">IOException { |
| 02 |
<code class="Brush keyword">super<code class="Brush plain">.setName(name); |
| 03 |
<code class="Brush keyword">this<code class="Brush plain">.port = port; |
| 04 |
<code class="Brush keyword">this<code class="Brush plain">.selector = Selector.open(); |
| 05 |
<code class="Brush keyword">this<code class="Brush plain">.serverChannel = ServerSocketChannel.open(); |
| 06 |
<code class="Brush comments">//ServerSocket使用TCP |
| 07 |
<code class="Brush keyword">this<code class="Brush plain">.serverChannel.socket().bind(<code class="Brush keyword">new <code class="Brush plain">InetSocketAddress(port)); |
| 08 |
<code class="Brush keyword">this<code class="Brush plain">.serverChannel.configureBlocking(<code class="Brush keyword">false<code class="Brush plain">); |
| 09 |
<code class="Brush keyword">this<code class="Brush plain">.serverChannel.register(selector, SelectionKey.OP_ACCEPT); |
| 10 |
<code class="Brush keyword">this<code class="Brush plain">.factory = factory; |
2、 然后会启动该线程，线程的run函数如下：
| 01 |
public <code class="Brush keyword">void <code class="Brush plain">run() { |
| 02 |
<code class="Brush keyword">final <code class="Brush plain">Selector selector = <code class="Brush keyword">this<code class="Brush plain">.selector; |
| 03 |
<code class="Brush comments">//线程一直循环 |
| 04 |
<code class="Brush keyword">for <code class="Brush plain">(;;) { |
| 05 |
<code class="Brush plain">++acceptCount; |
| 06 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 07 |
<code class="Brush plain">selector.select(1000L); |
| 08 |
<code class="Brush plain">Set<selectionkey> keys = selector.selectedKeys();</selectionkey> |
| 09 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 10 |
<code class="Brush keyword">for <code class="Brush plain">(SelectionKey key : keys) { |
| 11 |
<code class="Brush keyword">if <code class="Brush plain">(key.isValid() && key.isAcceptable()) { |
| 12 |
<code class="Brush comments">//接受来自客户端的连接 |
| 13 |
<code class="Brush plain">accept(); |
| 14 |
<code class="Brush plain">} <code class="Brush keyword">else <code class="Brush plain">{ |
| 15 |
<code class="Brush plain">key.cancel(); |
| 16 |
<code class="Brush plain">} |
| 17 |
<code class="Brush plain">} |
| 18 |
<code class="Brush plain">} <code class="Brush keyword">finally <code class="Brush plain">{ |
| 19 |
<code class="Brush plain">keys.clear(); |
| 20 |
<code class="Brush plain">} |
| 21 |
<code class="Brush plain">} <code class="Brush keyword">catch <code class="Brush plain">(Throwable e) { |
| 22 |
<code class="Brush plain">LOGGER.warn(getName(), e); |
| 23 |
<code class="Brush plain">} |
| 24 |
<code class="Brush plain">} |
3 、 该线程会一直循环监听想该selector注册过的server channel所感兴趣的事件（OP_ACCEPT），当有新的连接请求时，selector就会返回，keys就是请求连接的所有的包含channel的key集合。
SelectionKey有如下属性：
- interest集合(使用&操作SelectionKey.OP_ACCEPT和key.interestOps())
- ready集合(key.readyOps()，可以使用&操作检测该集合，也可以使用is方法)
- Channel(key.channel())
- Selector(key.selector())
- 附加对象(key.attach(obj) Object obj = key.attachment())
4、 然后遍历该集合，如果集合中的key没有被cancel，并且这个key的channel已经做好接受一个新的socket连接的准备，则接受该连接。
accept()的具体代码如下：
| 01 |
private <code class="Brush keyword">void <code class="Brush plain">accept() { |
| 02 |
<code class="Brush plain">SocketChannel channel = <code class="Brush keyword">null<code class="Brush plain">; |
| 03 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 04 |
<code class="Brush comments">//从服务器端获取管道,为一个新的连接返回channel |
| 05 |
<code class="Brush plain">channel = serverChannel.accept(); |
| 06 |
<code class="Brush comments">//配置管道为非阻塞 |
| 07 |
<code class="Brush plain">channel.configureBlocking(<code class="Brush keyword">false<code class="Brush plain">); |
| 09 |
<code class="Brush comments">//前端连接工厂对管道进行配置，设置socket的收发缓冲区大小，TCP延迟等 |
| 10 |
<code class="Brush comments">//然后由成员变量factory的类型生产对于的类型的连接 |
| 11 |
<code class="Brush comments">//比如ServerConnectionFactory会返回ServerConnection实例，并对其属性进行设置 |
| 12 |
<code class="Brush plain">FrontendConnection c = factory.make(channel); |
| 13 |
<code class="Brush comments">//设置连接属性 |
| 14 |
<code class="Brush plain">c.setAccepted(<code class="Brush keyword">true<code class="Brush plain">); |
| 15 |
<code class="Brush plain">c.setId(ID_GENERATOR.getId()); |
| 16 |
<code class="Brush comments">//从processors中选择一个NIOProcessor，将其和该连接绑定 |
| 17 |
<code class="Brush plain">NIOProcessor processor = nextProcessor(); |
| 18 |
<code class="Brush plain">c.setProcessor(processor); |
| 19 |
<code class="Brush comments">//向读反应堆注册该连接，加入待处理队列 |
| 20 |
<code class="Brush comments">//select选择到感兴趣的事件后，会进行调用connection的read函数 |
| 21 |
<code class="Brush plain">processor.postRegister(c); |
| 22 |
<code class="Brush plain">} <code class="Brush keyword">catch <code class="Brush plain">(Throwable e) { |
| 23 |
<code class="Brush plain">closeChannel(channel); |
| 24 |
<code class="Brush plain">LOGGER.warn(getName(), e); |
| 25 |
<code class="Brush plain">} |
| 26 |
<code class="Brush plain">} |
首先从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线程运行代码：
| 01 |
public <code class="Brush keyword">void <code class="Brush plain">run() { |
| 02 |
<code class="Brush keyword">final <code class="Brush plain">Selector selector = <code class="Brush keyword">this<code class="Brush plain">.selector; |
| 03 |
<code class="Brush keyword">for <code class="Brush plain">(;;) { |
| 04 |
<code class="Brush plain">++reactCount; |
| 05 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 06 |
<code class="Brush keyword">int <code class="Brush plain">res = selector.select(); |
| 07 |
<code class="Brush plain">LOGGER.debug(reactCount + <code class="Brush string">">>NIOReactor接受连接数:" <code class="Brush plain">+ res); |
| 08 |
<code class="Brush plain">register(selector); |
| 09 |
<code class="Brush plain">Set<selectionkey> keys = selector.selectedKeys();</selectionkey> |
| 10 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 11 |
<code class="Brush keyword">for <code class="Brush plain">(SelectionKey key : keys) { |
| 12 |
<code class="Brush plain">Object att = key.attachment(); |
| 13 |
<code class="Brush keyword">if <code class="Brush plain">(att != <code class="Brush keyword">null <code class="Brush plain">&& key.isValid()) { |
| 14 |
<code class="Brush keyword">int <code class="Brush plain">readyOps = key.readyOps(); |
| 15 |
<code class="Brush keyword">if <code class="Brush plain">((readyOps & SelectionKey.OP_READ) != <code class="Brush value">0<code class="Brush plain">) { |
| 16 |
<code class="Brush plain">LOGGER.debug(<code class="Brush string">"select读事件"<code class="Brush plain">); |
| 17 |
<code class="Brush plain">read((NIOConnection) att); |
| 18 |
<code class="Brush plain">} else if ((readyOps & SelectionKey.OP_WRITE) != 0) { |
| 19 |
<code class="Brush plain">LOGGER.debug(<code class="Brush string">"select写事件"<code class="Brush plain">); |
| 20 |
<code class="Brush plain">write((NIOConnection) att); |
| 21 |
<code class="Brush plain">} <code class="Brush keyword">else <code class="Brush plain">{ |
| 22 |
<code class="Brush plain">key.cancel(); |
| 23 |
<code class="Brush plain">} |
| 24 |
<code class="Brush plain">} <code class="Brush keyword">else <code class="Brush plain">{ |
| 25 |
<code class="Brush plain">key.cancel(); |
| 26 |
<code class="Brush plain">} |
| 27 |
<code class="Brush plain">} |
| 28 |
<code class="Brush plain">} <code class="Brush keyword">finally <code class="Brush plain">{ |
| 29 |
<code class="Brush plain">keys.clear(); |
| 30 |
<code class="Brush plain">} |
| 31 |
<code class="Brush plain">} <code class="Brush keyword">catch <code class="Brush plain">(Throwable e) { |
| 32 |
<code class="Brush plain">LOGGER.warn(name, e); |
| 33 |
<code class="Brush plain">} |
| 34 |
<code class="Brush plain">} |
该R线程也会一直循环运行，如果向该selector注册过的channel没有对应的感兴趣的事件发生，就会阻塞，直到有感兴趣的事件发生或被wakeup。返回后会运行register函数，将之前加入该reactor连接队列中的所有连接向该selector注册OP_READ事件。该注册的动作会调用Connection对象中的register方法进行注册
channel.register(selector, SelectionKey.OP_READ, this);
注意最后一个this指针参数，表示将该连接作为附件，注册到selector，当有感兴趣的时间发生时，函数selector.selectedKeys()返回的SelectionKey集合中的对象中使用key.attachment()即可获取到上面注册时绑定的connection对象指针附件。目的就是为了通过该附件对象调用该连接类中定义的read函数来完成功能。如下所示：
| 1 |
private <code class="Brush keyword">void <code class="Brush plain">read(NIOConnection c) { |
| 2 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 3 |
<code class="Brush plain">c.read(); |
| 4 |
<code class="Brush plain">} catch (Throwable e) { |
| 5 |
<code class="Brush plain">c.error(ErrorCode.ERR_READ, e); |
| 6 |
<code class="Brush plain">} |
6、 连接类中定义的read函数定义在AbstractConnection类中。在该read函数（该read函数涉及到的逻辑比较复杂，先不深究）中，完成从channel中读取数据到buffer，然后从buffer中提取byte数据交给具体子类（FrontendConnection）的handle()方法进行处理。
7、 该方法会从processor的线程池中获取一个线程，来异步执行数据的处理。处理会调用成员handler的handle方法来对数据进行处理。这里，在FrontendConnection的构造函数中定handler设置为FrontendAuthenticator（进行前端认证）。
| 01 |
public <code class="Brush keyword">void <code class="Brush plain">handle(<code class="Brush keyword">final <code class="Brush keyword">byte<code class="Brush plain">[] data) { |
| 02 |
<code class="Brush comments">// 从线程池获取一个线程,异步处理前端数据 |
| 03 |
<code class="Brush comments">// 从processor中的线程池中获取一个可以执行的线程,执行Runnable任务 |
| 04 |
<code class="Brush plain">processor.getHandler().execute(<code class="Brush keyword">new <code class="Brush plain">Runnable() { |
| 05 |
<code class="Brush color1">@Override |
| 06 |
public <code class="Brush keyword">void <code class="Brush plain">run() { |
| 07 |
<code class="Brush keyword">try <code class="Brush plain">{ |
| 08 |
<code class="Brush comments">//调用具体NIOHandler子类的handle函数 |
| 09 |
<code class="Brush plain">handler.handle(data); |
| 10 |
<code class="Brush plain">} catch (Throwable t) { |
| 11 |
<code class="Brush plain">error(ErrorCode.ERR_HANDLE_DATA, t); |
| 12 |
<code class="Brush plain">} |
| 13 |
<code class="Brush plain">} |
| 14 |
<code class="Brush plain">}); |
8、 handler在构造函数中初始化成前端认证处理器，用于处理前端权限认证。
| 1 |
public <code class="Brush plain">FrontendConnection(SocketChannel channel) { |
| 2 |
<code class="Brush keyword">super<code class="Brush plain">(channel); |
| 3 |
<code class="Brush plain">..................... |
| 4 |
<code class="Brush comments">//前端认证处理器 |
| 5 |
<code class="Brush keyword">this<code class="Brush plain">.handler = <code class="Brush keyword">new <code class="Brush plain">FrontendAuthenticator(<code class="Brush keyword">this<code class="Brush plain">); |
9、 由于Cobar是基于MySQL协议的，所以需要分析一下MySQL协议的具体格式。下面就先分析一下MySQL认证数据包的格式：
每个报文都分为消息头和消息体两部分，其中消息头是固定的四个字节，报文结构如下：
登录认证报文的报文数据部分格式如下：
10、 FrontendAuthenticator类对上面的数据包的具体处理如下：
- 读取信息到认证包对象
- 核对用户
- 核对密码
- 检查schema
如果出现错误，会提示相应的错误信息，如果正确会向客户端发送认证成功提示。
| 01 |
public <code class="Brush keyword">void <code class="Brush plain">handle(<code class="Brush keyword">byte<code class="Brush plain">[] data) { |
| 02 |
<code class="Brush comments">// check quit packet |
| 03 |
<code class="Brush keyword">if <code class="Brush plain">(data.length == QuitPacket.QUIT.length && data[<code class="Brush value">4<code class="Brush plain">] == MySQLPacket.COM_QUIT) { |
| 04 |
<code class="Brush plain">source.close(); |
| 05 |
<code class="Brush keyword">return<code class="Brush plain">; |
| 06 |
<code class="Brush plain">} |
| 08 |
<code class="Brush comments">//新建认证包对象 |
| 09 |
<code class="Brush plain">AuthPacket auth = <code class="Brush keyword">new <code class="Brush plain">AuthPacket(); |
| 10 |
<code class="Brush comments">//读取认证包到对象 |
| 11 |
<code class="Brush plain">auth.read(data); |
| 12 |
<code class="Brush comments">// check user |
| 13 |
<code class="Brush keyword">if <code class="Brush plain">(!checkUser(auth.user, source.getHost())) { |
| 14 |
<code class="Brush plain">failure(ErrorCode.ER_ACCESS_DENIED_ERROR, <code class="Brush string">"Access denied for user '" <code class="Brush plain">+ auth.user + <code class="Brush string">"'"<code class="Brush plain">); |
| 15 |
<code class="Brush keyword">return<code class="Brush plain">; |
| 16 |
<code class="Brush plain">} |
| 17 |
<code class="Brush comments">// check password |
| 18 |
<code class="Brush keyword">if <code class="Brush plain">(!checkPassword(auth.password, auth.user)) { |
| 19 |
<code class="Brush plain">failure(ErrorCode.ER_ACCESS_DENIED_ERROR, <code class="Brush string">"Access denied for user '" <code class="Brush plain">+ auth.user + <code class="Brush string">"'"<code class="Brush plain">); |
| 20 |
<code class="Brush keyword">return<code class="Brush plain">; |
| 21 |
<code class="Brush plain">} |
| 22 |
<code class="Brush comments">// check schema |
| 23 |
<code class="Brush keyword">switch <code class="Brush plain">(checkSchema(auth.database, auth.user)) { |
| 24 |
<code class="Brush keyword">case <code class="Brush plain">ErrorCode.ER_BAD_DB_ERROR: |
| 25 |
<code class="Brush plain">failure(ErrorCode.ER_BAD_DB_ERROR, <code class="Brush string">"Unknown database '" <code class="Brush plain">+ auth.database + <code class="Brush string">"'"<code class="Brush plain">); |
| 26 |
<code class="Brush keyword">break<code class="Brush plain">; |
| 27 |
<code class="Brush keyword">case <code class="Brush plain">ErrorCode.ER_DBACCESS_DENIED_ERROR: |
| 28 |
<code class="Brush plain">String s = <code class="Brush string">"Access denied for user '" <code class="Brush plain">+ auth.user + <code class="Brush string">"' to database '" <code class="Brush plain">+ auth.database + <code class="Brush string">"'"<code class="Brush plain">; |
| 29 |
<code class="Brush plain">failure(ErrorCode.ER_DBACCESS_DENIED_ERROR, s); |
| 30 |
<code class="Brush keyword">break<code class="Brush plain">; |
| 31 |
<code class="Brush keyword">default<code class="Brush plain">: |
| 32 |
<code class="Brush comments">//认证成功,向客户端发送认证结果消息 |
| 33 |
<code class="Brush plain">success(auth); |
| 34 |
<code class="Brush plain">} |
在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。认证成功后会执行：
| 01 |
protected <code class="Brush keyword">void <code class="Brush plain">success(AuthPacket auth) { |
| 02 |
<code class="Brush comments">//认证通过,设置连接属性:已认证\用户\数据库\处理器 |
| 03 |
<code class="Brush plain">source.setAuthenticated(<code class="Brush keyword">true<code class="Brush plain">); |
| 04 |
<code class="Brush plain">source.setUser(auth.user); |
| 05 |
<code class="Brush plain">source.setSchema(auth.database); |
| 06 |
<code class="Brush plain">source.setCharsetIndex(auth.charsetIndex); |
| 07 |
<code class="Brush comments">//设置该连接的连接处理器为前端命令处理器 |
| 08 |
<code class="Brush plain">source.setHandler(<code class="Brush keyword">new <code class="Brush plain">FrontendCommandHandler(source)); |
| 09 |
<code class="Brush plain">....... |
| 10 |
<code class="Brush plain">ByteBuffer buffer = source.allocate(); |
| 11 |
<code class="Brush plain">source.write(source.writeToBuffer(AUTH_OK, buffer)); |
可以看到，在上面的函数中，设置连接对象source中的成员（是否认证、用户、数据库、编码、处理该连接后续数据包的处理器【handle方法】）
然后回复认证成功的消息。后面客户端再发送消息，会交给前端命令处理器进行处理。
客户端进行链接的时候Cobar服务器的输出：
| 01 |
16:59:19,388 INFO =============================================== |
| 02 |
16:59:19,389 INFO Cobar is ready to startup ... |
| 03 |
16:59:19,389 INFO Startup processors ... |
| 04 |
16:59:19,455 INFO Startup connector ... |
| 05 |
16:59:19,460 INFO Initialize dataNodes ... |
| 06 |
16:59:19,506 INFO dnTest1:0 init success |
| 07 |
16:59:19,514 INFO dnTest3:0 init success |
| 08 |
16:59:19,517 INFO dnTest2:0 init success |
| 09 |
16:59:19,527 INFO CobarServer is started and listening on 8066 |
| 10 |
16:59:19,527 INFO =============================================== |
| 11 |
16:59:23,459 DEBUG 1>>NIOReactor接受连接数:0 |
| 12 |
16:59:23,464 DEBUG 2>>NIOReactor接受连接数:1 |
| 13 |
16:59:23,465 DEBUG select读事件 |
| 14 |
16:59:23,465 INFO com.alibaba.cobar.net.handler.FrontendAuthenticator接收的请求长度:62 |
| 15 |
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 |
16:59:23,468 INFO [thread=Processor1-H0,class=ServerConnection,host=192.168.137.8,port=46101,schema=null]'root' login success |
客户端得到的回复：
| 01 |
yan@yan-Z400:~$ mysql -uroot -p** -P8066 -h192.168.137.8 |
| 02 |
Welcome to the MySQL monitor. Commands end with ; or \g. |
| 03 |
Your MySQL connection id is 1 |
| 04 |
Server version: 5.1.48-cobar-1.2.7 Cobar Server (ALIBABA) |
| 06 |
Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved. |
| 08 |
Oracle is a registered trademark of Oracle Corporation and/or its |
| 09 |
affiliates. Other names may be trademarks of their respective |
| 12 |
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. |
MySQL客户端的命令处理，具体后续会分析。
作者：GeekCome出处：极客来原文：分布式数据库中间件–(2) Cobar与客户端的握手认证提示：本文版权归作者，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接。如果对文章有任何问题，都可以在评论中留言，我会尽可能的答复您，谢谢你的阅读
（完）
