分布式数据库中间件–(2) Cobar与client握手身份验证

Cobar启动完毕，监听特定端口。整个认证的流程图：

NIOAcceptor类继承自Thread类，该类的对象会以线程的方式执行，进行连接的监听。

NIOAcceptor启动的初始化步骤例如以下：

1 、打开一个selector，获取一个ServerSocketChannel对象。对该对象的socket绑定特定的监听端口，并设置该channel为非堵塞模式，然后想selector注冊该channel，绑定感兴趣的事件位OP_ACCEPT。

`01`	`public` `NIOAcceptor(String name,` `int` `port, FrontendConnectionFactory factory)` `throws` `IOException {`

`02`	`super.setName(name);`

`03`	`this.port = port;`

`04`	`this.selector = Selector.open();`

`05`	`this.serverChannel = ServerSocketChannel.open();`

`06`	`//ServerSocket使用TCP`

`07`	`this.serverChannel.socket().bind(new` `InetSocketAddress(port));`

`08`	`this.serverChannel.configureBlocking(false);`

`09`	`this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);`

`10`	`this.factory = factory;`

11 }

2、然后会启动该线程，线程的run函数例如以下：

`01`	`public` `void` `run() {`

`02`	`final` `Selector selector =` `this.selector;`

`03`	`//线程一直循环`

`04`	`for` `(;;) {`

`05`	`++acceptCount;`

06 try {

`07`	`selector.select(1000L);`

`08`	`Set<SelectionKey> keys = selector.selectedKeys();`

09 try {

`10`	`for` `(SelectionKey key : keys) {`

`11`	`if` `(key.isValid() && key.isAcceptable()) {`

`12`	`//接受来自client的连接`

`13`	`accept();`

`14`	`}` `else` `{`

`15`	`key.cancel();`

16 }

17 }

`18`	`}` `finally` `{`

`19`	`keys.clear();`

20 }

`21`	`}` `catch` `(Throwable e) {`

`22`	`LOGGER.warn(getName(), e);`

23 }

24 }

25 }

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` `void` `accept() {`

`02`	`SocketChannel channel =` `null;`

03 try {

`04`	`//从服务器端获取管道,为一个新的连接返回channel`

`05`	`channel = serverChannel.accept();`

`06`	`//配置管道为非堵塞`

`07`	`channel.configureBlocking(false);`

08

`09`	`//前端连接工厂对管道进行配置，设置socket的收发缓冲区大小，TCP延迟等`

`10`	`//然后由成员变量factory的类型生产对于的类型的连接`

`11`	`//比方ServerConnectionFactory会返回ServerConnection实例，并对其属性进行设置`

`12`	`FrontendConnection c = factory.make(channel);`

`13`	`//设置连接属性`

`14`	`c.setAccepted(true);`

`15`	`c.setId(ID_GENERATOR.getId());`

`16`	`//从processors中选择一个NIOProcessor。将其和该连接绑定`

`17`	`NIOProcessor processor = nextProcessor();`

`18`	`c.setProcessor(processor);`

`19`	`//向读反应堆注冊该连接，增加待处理队列`

`20`	`//select选择到感兴趣的事件后，会进行调用connection的read函数`

`21`	`processor.postRegister(c);`

`22`	`}` `catch` `(Throwable e) {`

`23`	`closeChannel(channel);`

`24`	`LOGGER.warn(getName(), e);`

25 }

26 }

首先从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` `void` `run() {`

`02`	`final` `Selector selector =` `this.selector;`

`03`	`for` `(;;) {`

`04`	`++reactCount;`

05 try {

`06`	`int` `res = selector.select();`

`07`	`LOGGER.debug(reactCount +` `">>NIOReactor接受连接数:"` `+ res);`

`08`	`register(selector);`

`09`	`Set<SelectionKey> keys = selector.selectedKeys();`

10 try {

`11`	`for` `(SelectionKey key : keys) {`

`12`	`Object att = key.attachment();`

`13`	`if` `(att !=` `null` `&& key.isValid()) {`

`14`	`int` `readyOps = key.readyOps();`

`15`	`if` `((readyOps & SelectionKey.OP_READ) !=` `0) {`

`16`	`LOGGER.debug("select读事件");`

`17`	`read((NIOConnection) att);`

`18`	`}` `else` `if` `((readyOps & SelectionKey.OP_WRITE) !=` `0) {`

`19`	`LOGGER.debug("select写事件");`

`20`	`write((NIOConnection) att);`

`21`	`}` `else` `{`

`22`	`key.cancel();`

23 }

`24`	`}` `else` `{`

`25`	`key.cancel();`

26 }

27 }

`28`	`}` `finally` `{`

`29`	`keys.clear();`

30 }

`31`	`}` `catch` `(Throwable e) {`

`32`	`LOGGER.warn(name, e);`

33 }

34 }

35 }

该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` `void` `read(NIOConnection c) {`

2 try {

`3`	`c.read();`

`4`	`}` `catch` `(Throwable e) {`

`5`	`c.error(ErrorCode.ERR_READ, e);`

6 }

7 }

6、连接类中定义的read函数定义在AbstractConnection类中。

在该read函数~~（该read函数涉及到的逻辑比較复杂。先不深究）~~中，完毕从channel中读取数据到buffer，然后从buffer中提取byte数据交给详细子类（FrontendConnection）的handle()方法进行处理。

7、该方法会从processor的线程池中获取一个线程，来异步执行数据的处理。处理会调用成员handler的handle方法来对数据进行处理。这里，在FrontendConnection的构造函数中定handler设置为FrontendAuthenticator（进行前端认证）。

`01`	`public` `void` `handle(final` `byte[] data) {`

`02`	`// 从线程池获取一个线程,异步处理前端数据`

`03`	`// 从processor中的线程池中获取一个能够执行的线程,执行Runnable任务`

`04`	`processor.getHandler().execute(new` `Runnable() {`

`05`	`@Override`

`06`	`public` `void` `run() {`

07 try {

`08`	`//调用详细NIOHandler子类的handle函数`

`09`	`handler.handle(data);`

`10`	`}` `catch` `(Throwable t) {`

`11`	`error(ErrorCode.ERR_HANDLE_DATA, t);`

12 }

13 }

14 });

15 }

8、 handler在构造函数中初始化成前端认证处理器。用于处理前端权限认证。

`1`	`public` `FrontendConnection(SocketChannel channel) {`

`2`	`super(channel);`

`3`	`.....................`

`4`	`//前端认证处理器`

`5`	`this.handler =` `new` `FrontendAuthenticator(this);`

6 }

9、因为Cobar是基于MySQL协议的，所以须要分析一下MySQL协议的详细格式。以下就先分析一下MySQL认证数据包的格式：

每一个报文都分为消息头和消息体两部分，当中消息头是固定的四个字节，报文结构例如以下：

登录认证报文的报文数据部分格式例如以下：

10、 FrontendAuthenticator类对上面的数据包的详细处理例如以下：

读取信息到认证包对象
核对用户
核对密码
检查schema

假设出现错误。会提示相应的错误信息，假设正确会向client发送认证成功提示。

`01`	`public` `void` `handle(byte[] data) {`

`02`	`// check quit packet`

`03`	`if` `(data.length == QuitPacket.QUIT.length && data[4] == MySQLPacket.COM_QUIT) {`

`04`	`source.close();`

05 return;

06 }

07

`08`	`//新建认证包对象`

`09`	`AuthPacket auth =` `new` `AuthPacket();`

`10`	`//读取认证包到对象`

`11`	`auth.read(data);`

`12`	`// check user`

`13`	`if` `(!checkUser(auth.user, source.getHost())) {`

`14`	`failure(ErrorCode.ER_ACCESS_DENIED_ERROR,` `"Access denied for user '"` `+ auth.user +` `"'");`

15 return;

16 }

`17`	`// check password`

`18`	`if` `(!checkPassword(auth.password, auth.user)) {`

`19`	`failure(ErrorCode.ER_ACCESS_DENIED_ERROR,` `"Access denied for user '"` `+ auth.user +` `"'");`

20 return;

21 }

`22`	`// check schema`

`23`	`switch` `(checkSchema(auth.database, auth.user)) {`

`24`	`case` `ErrorCode.ER_BAD_DB_ERROR:`

`25`	`failure(ErrorCode.ER_BAD_DB_ERROR,` `"Unknown database '"` `+ auth.database +` `"'");`

26 break;

`27`	`case` `ErrorCode.ER_DBACCESS_DENIED_ERROR:`

`28`	`String s =` `"Access denied for user '"` `+ auth.user +` `"' to database '"` `+ auth.database +` `"'";`

`29`	`failure(ErrorCode.ER_DBACCESS_DENIED_ERROR, s);`

30 break;

`31`	`default:`

`32`	`//认证成功,向client发送认证结果消息`

`33`	`success(auth);`

34 }

35 }

在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。

认证成功后会执行：

`01`	`protected` `void` `success(AuthPacket auth) {`

`02`	`//认证通过,设置连接属性:已认证\用户\数据库\处理器`

`03`	`source.setAuthenticated(true);`

`04`	`source.setUser(auth.user);`

`05`	`source.setSchema(auth.database);`

`06`	`source.setCharsetIndex(auth.charsetIndex);`

`07`	`//设置该连接的连接处理器为前端命令处理器`

`08`	`source.setHandler(new` `FrontendCommandHandler(source));`

09 .......

`10`	`ByteBuffer buffer = source.allocate();`

`11`	`source.write(source.writeToBuffer(AUTH_OK, buffer));`

12 }

能够看到，在上面的函数中，设置连接对象source中的成员（是否认证、用户、数据库、编码、处理该连接兴许数据包的处理器【handle方法】）

然后回复认证成功的消息。

后面client再发送消息，会交给前端命令处理器进行处理。

client进行链接的时候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`

client得到的回复：

`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)`

05

`06`	`Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.`

07

`08`	`Oracle is a registered trademark of Oracle Corporation and/or its`

`09`	`affiliates. Other names may be trademarks of their respective`

10 owners.

11

`12`	`Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.`

13

14 mysql>

MySQLclient的命令处理。详细兴许会分析。

作者：GeekCome
出处：极客来
原文：分布式数据库中间件–(2) Cobar与client的握手认证
提示：本文版权归作者。欢迎转载。但未经作者允许必须保留此段声明，且在文章页面明显位置给出原文连接。
假设对文章有不论什么问题，都能够在评论中留言。我会尽可能的答复您，谢谢你的阅读

（完）