分布式数据库中间件–(2)Cobar与客户端的握手认证-Mysql Tutorial-php.cn

Cobar启动完成，监听特定端口。整个认证的流程图： NIOAcceptor类继承自Thread类，该类的对象会以线程的方式运行，进行连接的监听。 NIOAcceptor启动的初始化过程如下： 1 、打开一个selector，获取一个ServerSocketChannel对象，对该对象的socket绑定特定的

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 keys = selector.selectedKeys();`

09 try {

10	`for (SelectionKey key : keys) {`

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

12	`//接受来自客户端的连接`

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 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
  如果出现错误，会提示相应的错误信息，如果正确会向客户端发送认证成功提示。
  
  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 //认证成功,向客户端发送认证结果消息
  
  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方法】）
  
  然后回复认证成功的消息。后面客户端再发送消息，会交给前端命令处理器进行处理。
  
  客户端进行链接的时候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)
  
  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>
  
  MySQL客户端的命令处理，具体后续会分析。
  作者：GeekCome出处：极客来原文：分布式数据库中间件–(2) Cobar与客户端的握手认证提示：本文版权归作者，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接。如果对文章有任何问题，都可以在评论中留言，我会尽可能的答复您，谢谢你的阅读
  
  （完）