cobar启动完成，监听特定端口。整个认证的流程图： nioacceptor类继承自thread类，该类的对象会以线程的方式运行，进行连接的监听。nioacceptor启动的初始化过程如下：1 、打开一个selector，获取一个serversocketchannel对象，对该对象的socket绑定特定的监
cobar启动完成，监听特定端口。整个认证的流程图：nioacceptor类继承自thread类，该类的对象会以线程的方式运行，进行连接的监听。nioacceptor启动的初始化过程如下：1 、打开一个selector，获取一个serversocketchannel对象，对该对象的socket绑定特定的监听端口，并设置该channel为非阻塞模式，然后想selector注册该channel，绑定感兴趣的事件位op_accept。 public nioacceptor(string name, int port, frontendconnectionfactory factory) throws ioexception { super.setname(name); this.port = port; this.selector = selector.open(); this.serverchannel = serversocketchannel.open(); //serversocket使用tcp this.serverchannel.socket().bind(new inetsocketaddress(port)); this.serverchannel.configureblocking(false); this.serverchannel.register(selector, selectionkey.op_accept); this.factory = factory; }
2、 然后会启动该线程，线程的run函数如下： public void run() { final selector selector = this.selector; //线程一直循环 for (;;) { ++acceptcount; try { selector.select(1000l); set keys = selector.selectedkeys(); try { for (selectionkey key : keys) { if (key.isvalid() && key.isacceptable()) { //接受来自客户端的连接 accept(); } else { key.cancel(); } } } finally { keys.clear(); } } catch (throwable e) { 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()的具体代码如下：private void accept() { socketchannel channel = null; try { //从服务器端获取管道,为一个新的连接返回channel channel = serverchannel.accept(); //配置管道为非阻塞 channel.configureblocking(false); //前端连接工厂对管道进行配置，设置socket的收发缓冲区大小，tcp延迟等 //然后由成员变量factory的类型生产对于的类型的连接 //比如serverconnectionfactory会返回serverconnection实例，并对其属性进行设置 frontendconnection c = factory.make(channel); //设置连接属性 c.setaccepted(true); c.setid(id_generator.getid()); //从processors中选择一个nioprocessor，将其和该连接绑定 nioprocessor processor = nextprocessor(); c.setprocessor(processor); //向读反应堆注册该连接，加入待处理队列 //select选择到感兴趣的事件后，会进行调用connection的read函数 processor.postregister(c); } catch (throwable e) { closechannel(channel); 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线程运行代码： public void run() { final selector selector = this.selector; for (;;) { ++reactcount; try { int res = selector.select(); logger.debug(reactcount + >>nioreactor接受连接数: + res); register(selector); set keys = selector.selectedkeys(); try { for (selectionkey key : keys) { object att = key.attachment(); if (att != null && key.isvalid()) { int readyops = key.readyops(); if ((readyops & selectionkey.op_read) != 0) { logger.debug(select读事件); read((nioconnection) att); } else if ((readyops & selectionkey.op_write) != 0) { logger.debug(select写事件); write((nioconnection) att); } else { key.cancel(); } } else { key.cancel(); } } } finally { keys.clear(); } } catch (throwable e) { logger.warn(name, e); } } }
该r线程也会一直循环运行，如果向该selector注册过的channel没有对应的感兴趣的事件发生，就会阻塞，直到有感兴趣的事件发生或被wakeup。返回后会运行register函数，将之前加入该reactor连接队列中的所有连接向该selector注册op_read事件。该注册的动作会调用connection对象中的register方法进行注册，并像客户端主动发送握手数据包（handshakepacket）。?channel.register(selector, selectionkey.op_read, this);注意最后一个this指针参数，表示将该连接作为附件，注册到selector，当有感兴趣的时间发生时，函数selector.selectedkeys()返回的selectionkey集合中的对象中使用key.attachment()即可获取到上面注册时绑定的connection对象指针附件。目的就是为了通过该附件对象调用该连接类中定义的read函数来完成功能。如下所示： private void read(nioconnection c) { try { c.read(); } catch (throwable e) { c.error(errorcode.err_read, e); } }
6、 连接类中定义的read函数定义在abstractconnection类中。在该read函数（该read函数涉及到的逻辑比较复杂，先不深究）中，完成从channel中读取数据到buffer，然后从buffer中提取byte数据交给具体子类（frontendconnection）的handle()方法进行处理。7、 该方法会从processor的线程池中获取一个线程，来异步执行数据的处理。处理会调用成员handler的handle方法来对数据进行处理。这里，在frontendconnection的构造函数中定handler设置为frontendauthenticator（进行前端认证）。 public void handle(final byte[] data) { // 从线程池获取一个线程,异步处理前端数据 // 从processor中的线程池中获取一个可以执行的线程,执行runnable任务 processor.gethandler().execute(new runnable() { @override public void run() { try { //调用具体niohandler子类的handle函数 handler.handle(data); } catch (throwable t) { error(errorcode.err_handle_data, t); } } }); }
8、 handler在构造函数中初始化成前端认证处理器，用于处理前端权限认证。 public frontendconnection(socketchannel channel) { super(channel); ..................... //前端认证处理器 this.handler = new frontendauthenticator(this); }
9、 由于cobar是基于mysql协议的，所以需要分析一下mysql协议的具体格式。下面就先分析一下mysql认证数据包的格式：每个报文都分为消息头和消息体两部分，其中消息头是固定的四个字节，报文结构如下：登录认证报文的报文数据部分格式如下：10、 frontendauthenticator类对上面的数据包的具体处理如下： 读取信息到认证包对象 核对用户 核对密码 检查schema如果出现错误，会提示相应的错误信息，如果正确会向客户端发送认证成功提示。 public void handle(byte[] data) { // check quit packet if (data.length == quitpacket.quit.length && data[4] == mysqlpacket.com_quit) { source.close(); return; } //新建认证包对象 authpacket auth = new authpacket(); //读取认证包到对象 auth.read(data); // check user if (!checkuser(auth.user, source.gethost())) { failure(errorcode.er_access_denied_error, access denied for user ' + auth.user + '); return; } // check password if (!checkpassword(auth.password, auth.user)) { failure(errorcode.er_access_denied_error, access denied for user ' + auth.user + '); return; } // check schema switch (checkschema(auth.database, auth.user)) { case errorcode.er_bad_db_error: failure(errorcode.er_bad_db_error, unknown database ' + auth.database + '); break; case errorcode.er_dbaccess_denied_error: string s = access denied for user ' + auth.user + ' to database ' + auth.database + '; failure(errorcode.er_dbaccess_denied_error, s); break; default: //认证成功,向客户端发送认证结果消息 success(auth); } }
在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。认证成功后会执行： protected void success(authpacket auth) { //认证通过,设置连接属性:已认证\用户\数据库\处理器 source.setauthenticated(true); source.setuser(auth.user); source.setschema(auth.database); source.setcharsetindex(auth.charsetindex); //设置该连接的连接处理器为前端命令处理器 source.sethandler(new frontendcommandhandler(source)); ....... bytebuffer buffer = source.allocate(); source.write(source.writetobuffer(auth_ok, buffer)); }
可以看到，在上面的函数中，设置连接对象source中的成员（是否认证、用户、数据库、编码、处理该连接后续数据包的处理器【handle方法】）然后回复认证成功的消息。后面客户端再发送消息，会交给前端命令处理器进行处理。客户端进行链接的时候cobar服务器的输出：16:59:19,388 info ===============================================16:59:19,389 info cobar is ready to startup ...16:59:19,389 info startup processors ...16:59:19,455 info startup connector ...16:59:19,460 info initialize datanodes ...16:59:19,506 info dntest1:0 init success16:59:19,514 info dntest3:0 init success16:59:19,517 info dntest2:0 init success16:59:19,527 info cobarserver is started and listening on 806616:59:19,527 info ===============================================16:59:23,459 debug 1>>nioreactor接受连接数:016:59:23,464 debug 2>>nioreactor接受连接数:116:59:23,465 debug select读事件16:59:23,465 info com.alibaba.cobar.net.handler.frontendauthenticator接收的请求长度:6258 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:59:23,468 info [thread=processor1-h0,class=serverconnection,host=192.168.137.8,port=46101,schema=null]'root' login success
客户端得到的回复：yan@yan-z400:~$ mysql -uroot -p** -p8066 -h192.168.137.8welcome to the mysql monitor. commands end with ; or \g.your mysql connection id is 1server version: 5.1.48-cobar-1.2.7 cobar server (alibaba)copyright (c) 2000, 2013, oracle and/or its affiliates. all rights reserved.oracle is a registered trademark of oracle corporation and/or itsaffiliates. other names may be trademarks of their respectiveowners.type 'help;' or '\h' for help. type '\c' to clear the current input statement.mysql>
mysql客户端的命令处理，具体后续会分析。 本文出自：http://blog.geekcome.com, 原文地址：http://blog.geekcome.com/%e5%88%86%e5%b8%83%e5%bc%8f%e6%95%b0%e6%8d%ae%e5%ba%93%e4%b8%ad%e9%97%b4%e4%bb%b6-2-cobar%e4%b8%8e%e5%ae%a2%e6%88%b7%e7%ab%af%e7%9a%84%e6%8f%a1%e6%89%8b%e8%ae%a4%e8%af%81, 感谢原作者分享。