标签:数据库 中间件 分布式 reactor alibaba
Cobar启动完成,监听特定端口。整个认证的流程图:
NIOAcceptor类继承自Thread类,该类的对象会以线程的方式运行,进行连接的监听。
NIOAcceptor启动的初始化过程如下:
1 、打开一个selector,获取一个ServerSocketChannel对象,对该对象的socket绑定特定的监听端口,并设置该channel为非阻塞模式,然后想selector注册该channel,绑定感兴趣的事件位OP_ACCEPT。
01 | public NIOAcceptor(String name, int port, FrontendConnectionFactory factory) throws IOException { |
04 | this .selector = Selector.open(); |
05 | this .serverChannel = ServerSocketChannel.open(); |
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; |
2、 然后会启动该线程,线程的run函数如下:
02 | final Selector selector = this .selector; |
07 | selector.select(1000L); |
08 | Set<SelectionKey> keys = selector.selectedKeys(); |
10 | for (SelectionKey key : keys) { |
11 | if (key.isValid() && key.isAcceptable()) { |
21 | } catch (Throwable e) { |
22 | LOGGER.warn(getName(), e); |
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 ; |
05 | channel = serverChannel.accept(); |
07 | channel.configureBlocking( false ); |
12 | FrontendConnection c = factory.make(channel); |
15 | c.setId(ID_GENERATOR.getId()); |
17 | NIOProcessor processor = nextProcessor(); |
18 | c.setProcessor(processor); |
21 | processor.postRegister(c); |
22 | } catch (Throwable e) { |
23 | closeChannel(channel); |
24 | 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线程运行代码:
02 | final Selector selector = this .selector; |
06 | int res = selector.select(); |
07 | LOGGER.debug(reactCount + ">>NIOReactor接受连接数:" + res); |
09 | Set<SelectionKey> keys = selector.selectedKeys(); |
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); |
31 | } catch (Throwable e) { |
该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) { |
4 | } catch (Throwable e) { |
5 | c.error(ErrorCode.ERR_READ, e); |
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) { |
04 | processor.getHandler().execute( new Runnable() { |
10 | } catch (Throwable t) { |
11 | error(ErrorCode.ERR_HANDLE_DATA, t); |
8、 handler在构造函数中初始化成前端认证处理器,用于处理前端权限认证。
1 | public FrontendConnection(SocketChannel channel) { |
5 | this .handler = new FrontendAuthenticator( this ); |
9、 由于Cobar是基于MySQL协议的,所以需要分析一下MySQL协议的具体格式。下面就先分析一下MySQL认证数据包的格式:
每个报文都分为消息头和消息体两部分,其中消息头是固定的四个字节,报文结构如下:
登录认证报文的报文数据部分格式如下:
10、 FrontendAuthenticator类对上面的数据包的具体处理如下:
- 读取信息到认证包对象
- 核对用户
- 核对密码
- 检查schema
如果出现错误,会提示相应的错误信息,如果正确会向客户端发送认证成功提示。
01 | public void handle( byte [] data) { |
03 | if (data.length == QuitPacket.QUIT.length && data[ 4 ] == MySQLPacket.COM_QUIT) { |
09 | AuthPacket auth = new AuthPacket(); |
13 | if (!checkUser(auth.user, source.getHost())) { |
14 | failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user ‘" + auth.user + "‘" ); |
18 | if (!checkPassword(auth.password, auth.user)) { |
19 | failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user ‘" + auth.user + "‘" ); |
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 + "‘" ); |
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); |
在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。认证成功后会执行:
01 | protected void success(AuthPacket auth) { |
03 | source.setAuthenticated( true ); |
04 | source.setUser(auth.user); |
05 | source.setSchema(auth.database); |
06 | source.setCharsetIndex(auth.charsetIndex); |
08 | source.setHandler( new FrontendCommandHandler(source)); |
10 | ByteBuffer buffer = source.allocate(); |
11 | 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客户端的命令处理,具体后续会分析。
提示:本文版权归作者,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接。
如果对文章有任何问题,都可以在评论中留言,我会尽可能的答复您,谢谢你的阅读
(完)
分布式数据库中间件–(2) Cobar与客户端的握手认证
标签:数据库 中间件 分布式 reactor alibaba
原文地址:http://blog.csdn.net/geekcome/article/details/24597405