search

Home >Database >Mysql Tutorial >分布式数据库中间件–(2)Cobar与客户端的握手认证

分布式数据库中间件–(2)Cobar与客户端的握手认证

WBOY
WBOYOriginal
2016-06-07 15:55:511335browse

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;
11 }

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">}
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 <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">);
    08
    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">}
    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 <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">}
    7 }

    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">});
    15 }

    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">);
    6 }

    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">}
      07
      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">}
      35 }

      在上面的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));
      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与客户端的握手认证提示：本文版权归作者，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接。如果对文章有任何问题，都可以在评论中留言，我会尽可能的答复您，谢谢你的阅读



      （完）

Statement：
The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn
Previous article：altersystemkillsessionimmediateNext article：ksvcreate:Process(m001)creationfailed问题处理

Related articles

See more