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

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

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

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对象。

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

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认证数据包的格式:

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

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

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

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

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
出处:极客来
提示:本文版权归作者。欢迎转载。但未经作者允许必须保留此段声明,且在文章页面明显位置给出原文连接。
假设对文章有不论什么问题,都能够在评论中留言。我会尽可能的答复您,谢谢你的阅读

(完)

版权声明:本文博客原创文章。博客,未经同意,不得转载。

上一篇:windos多线程编程


下一篇:DIV周边添加投影及背景固定