文档章节

MyCat - 源代码篇(2)

张哈希_
 张哈希_
发布于 2016/04/24 11:41
字数 1705
阅读 11
收藏 0

数据库路由中间件MyCat - 源代码篇(2)

2. 前端连接建立与认证

Created with Raphaël 2.1.0 MySql连接建立以及认证过程 client client MySql MySql 1.TCP连接请求 2.接受TCP连接 3.TCP连接建立 4.握手包HandshakePacket 5.认证包AuthPacket 6.如果验证成功,则返回OkPacket 7.默认会发送查询版本信息的包 8.返回结果包

2.2 (4)握手包HandshakePacket

NIOReactor其实就是一个网络事件反应转发器。
这里写图片描述
很多地方会用到NIOReactor,这里先讲FrontendConnection和NIOReactor绑定这一部分。上一节说到,NIOAcceptor的accept()最后将FrontendConnection交给了NIOReactor池其中的一个NIOReactor。调用的是 postRegister(AbstractConnection c)方法。

final void postRegister(AbstractConnection c) {
        reactorR.registerQueue.offer(c);
        reactorR.selector.wakeup();
    }

postRegister将刚才传入的FrontendConnection放入RW线程的注册队列。之后,唤醒RW线程的selector。
为什么放入RW线程的注册队列,而不是直接注册呢?如果是直接注册,那么就是NIOAcceptor这个线程负责注册,这里就会有锁竞争,因为NIOAcceptor这个线程和每个RW线程会去竞争selector的锁。这样NIOAcceptor就不能高效的处理连接。所以,更好的方式是将FrontendConnection放入RW线程的注册队列,之后让RW线程自己完成注册工作。
RW线程的源代码:

private final class RW implements Runnable {
        private final Selector selector;
        private final ConcurrentLinkedQueue<AbstractConnection> registerQueue;
        private long reactCount;

        private RW() throws IOException {
            this.selector = Selector.open();
            this.registerQueue = new ConcurrentLinkedQueue<AbstractConnection>();
        }

        @Override
        public void run() {
            final Selector selector = this.selector;
            Set<SelectionKey> keys = null;
            for (;;) {
                ++reactCount;
                try {
                    selector.select(500L);
                    //从注册队列中取出AbstractConnection之后注册读事件
                    //之后做一些列操作,请参考下面注释
                    register(selector);
                    keys = selector.selectedKeys();
                    for (SelectionKey key : keys) {
                        AbstractConnection con = null;
                        try {
                            Object att = key.attachment();
                            if (att != null) {
                                con = (AbstractConnection) att;
                                if (key.isValid() && key.isReadable()) {
                                    try {
                                        //异步读取数据并处理数据
                                        con.asynRead();
                                    } catch (IOException e) {
                                        con.close("program err:" + e.toString());
                                        continue;
                                    } catch (Exception e) {
                                        LOGGER.debug("caught err:", e);
                                        con.close("program err:" + e.toString());
                                        continue;
                                    }
                                }
                                if (key.isValid() && key.isWritable()) {
                                    //异步写数据
                                    con.doNextWriteCheck();
                                }
                            } else {
                                key.cancel();
                            }
                        } catch (CancelledKeyException e) {
                            if (LOGGER.isDebugEnabled()) {
                                LOGGER.debug(con + " socket key canceled");
                            }
                        } catch (Exception e) {
                            LOGGER.warn(con + " " + e);
                        }
                    }
                } catch (Exception e) {
                    LOGGER.warn(name, e);
                } finally {
                    if (keys != null) {
                        keys.clear();
                    }

                }
            }
        }

        private void register(Selector selector) {
            AbstractConnection c = null;
            if (registerQueue.isEmpty()) {
                return;
            }
            while ((c = registerQueue.poll()) != null) {
                try {
                    //注册读事件
                    ((NIOSocketWR) c.getSocketWR()).register(selector);
                    //连接注册,对于FrontendConnection是发送HandshakePacket并异步读取响应
                    //响应为AuthPacket,读取其中的信息,验证用户名密码等信息,如果符合条件
                    //则发送OkPacket
                    c.register();
                } catch (Exception e) {
                    c.close("register err" + e.toString());
                }
            }
        }

    }

因为NIOAcceptor线程和RW线程这两个都会操作RW线程的注册队列,所以要用ConcurrentLinkedQueue
RW线程不断检查selector中需要响应的事件,并如果注册队列不为空,就不断注册其中的AbstractConnection,在这里就是FrontendConnection。
之后执行FrontendConnection的register()方法:

@Override
    public void register() throws IOException {
        if (!isClosed.get()) {

            // 生成认证数据
            byte[] rand1 = RandomUtil.randomBytes(8);
            byte[] rand2 = RandomUtil.randomBytes(12);

            // 保存认证数据
            byte[] seed = new byte[rand1.length + rand2.length];
            System.arraycopy(rand1, 0, seed, 0, rand1.length);
            System.arraycopy(rand2, 0, seed, rand1.length, rand2.length);
            this.seed = seed;

            // 发送握手数据包
            HandshakePacket hs = new HandshakePacket();
            hs.packetId = 0;
            hs.protocolVersion = Versions.PROTOCOL_VERSION;
            hs.serverVersion = Versions.SERVER_VERSION;
            hs.threadId = id;
            hs.seed = rand1;
            hs.serverCapabilities = getServerCapabilities();
            hs.serverCharsetIndex = (byte) (charsetIndex & 0xff);
            hs.serverStatus = 2;
            hs.restOfScrambleBuff = rand2;
            // 异步写,本节就讲到这里
            hs.write(this);

            // 异步读取并处理,这个与RW线程中的asynRead()相同,之后客户端收到握手包返回AuthPacket(就是下一节)就是从这里开始看。
            this.asynRead();
        }
    }

这个方法就是生成HandshakePacket并发送出去,之后异步读取响应。
之前的示例中MySql的HandshakePacket结构:
HandshakePacket
可以总结出:
HandshakePacket:
- packet length(3 bytes)
- packet number (1)
- protocol version (1)
- version (null terminated string)
- thread id (4)
- salt (8)
- server capabilities (2)
- server charset (1)
- server status (2)
- unused (13)
- salt (12)
- 0x00 — 结束

这里我们看下MyCat中的实现这一部分MySql协议栈的packet类结构:
这里写图片描述
这里可以看出,每个包都实现了自己的包长度和信息方法,并且针对前段后端连接都有读写方法实现,所以,之后读写数据都会根据场景不同调用这些类中的方法。这些包就是整个MySql协议栈除逻辑外的内容实现。
HandshakePacket.write(FrontendConnection c)方法将上面传入的数据封装成ByteBuffer,并传入给FrontendConnection c的write(ByteBuffer buffer),这个方法直接继承自AbstractConnection:

public final void write(ByteBuffer buffer) {
        //首先判断是否为压缩协议
        if(isSupportCompress())
        {
            //CompressUtil为压缩协议辅助工具类
            ByteBuffer     newBuffer= CompressUtil.compressMysqlPacket(buffer,this,compressUnfinishedDataQueue);
            //将要写的数据先放入写缓存队列
            writeQueue.offer(newBuffer);

        }   else
        {
            //将要写的数据先放入写缓存队列
            writeQueue.offer(buffer);
        }
        try {
            //处理写事件,这个方法比较复杂,需要重点分析其思路
            this.socketWR.doNextWriteCheck();
        } catch (Exception e) {
            LOGGER.warn("write err:", e);
            this.close("write err:" + e);

        }

    }

如代码注释中所述,先将要写的数据放入写缓冲队列,之后调用NIOSocketWR.doNextWriteCheck()处理写事件。

public void doNextWriteCheck() {
        //检查是否正在写,看CAS更新writing值是否成功
        if (!writing.compareAndSet(false, true)) {
            return;
        }

        try {
            //利用缓存队列和写缓冲记录保证写的可靠性,返回true则为全部写入成功
            boolean noMoreData = write0();
            //因为只有一个线程可以成功CAS更新writing值,所以这里不用再CAS
            writing.set(false);
            //如果全部写入成功而且写入队列为空(有可能在写入过程中又有新的Bytebuffer加入到队列),则取消注册写事件
            //否则,继续注册写事件
            if (noMoreData && con.writeQueue.isEmpty()) {
                if ((processKey.isValid() && (processKey.interestOps() & SelectionKey.OP_WRITE) != 0)) {
                    disableWrite();
                }

            } else {

                if ((processKey.isValid() && (processKey.interestOps() & SelectionKey.OP_WRITE) == 0)) {
                    enableWrite(false);
                }
            }

        } catch (IOException e) {
            if (AbstractConnection.LOGGER.isDebugEnabled()) {
                AbstractConnection.LOGGER.debug("caught err:", e);
            }
            con.close("err:" + e);
        }

    }

    private boolean write0() throws IOException {
        int written = 0;
        ByteBuffer buffer = con.writeBuffer;
        if (buffer != null) {
            //只要写缓冲记录中还有数据就不停写入,但如果写入字节为0,证明网络繁忙,则退出
            while (buffer.hasRemaining()) {
                written = channel.write(buffer);
                if (written > 0) {
                    con.netOutBytes += written;
                    con.processor.addNetOutBytes(written);
                    con.lastWriteTime = TimeUtil.currentTimeMillis();
                } else {
                    break;
                }
            }
            //如果写缓冲中还有数据证明网络繁忙,计数并退出,否则清空缓冲
            if (buffer.hasRemaining()) {
                con.writeAttempts++;
                return false;
            } else {
                con.writeBuffer = null;
                con.recycle(buffer);
            }
        }
        //读取缓存队列并写channel
        while ((buffer = con.writeQueue.poll()) != null) {
            if (buffer.limit() == 0) {
                con.recycle(buffer);
                con.close("quit send");
                return true;
            }
            buffer.flip();
            while (buffer.hasRemaining()) {
                written = channel.write(buffer);
                if (written > 0) {
                    con.lastWriteTime = TimeUtil.currentTimeMillis();
                    con.netOutBytes += written;
                    con.processor.addNetOutBytes(written);
                    con.lastWriteTime = TimeUtil.currentTimeMillis();
                } else {
                    break;
                }
            }
            //如果写缓冲中还有数据证明网络繁忙,计数,记录下这次未写完的数据到写缓冲记录并退出,否则回收缓冲
            if (buffer.hasRemaining()) {
                con.writeBuffer = buffer;
                con.writeAttempts++;
                return false;
            } else {
                con.recycle(buffer);
            }
        }
        return true;
    }

    private void disableWrite() {
        try {
            SelectionKey key = this.processKey;
            key.interestOps(key.interestOps() & OP_NOT_WRITE);
        } catch (Exception e) {
            AbstractConnection.LOGGER.warn("can't disable write " + e + " con "
                    + con);
        }

    }

    private void enableWrite(boolean wakeup) {
        boolean needWakeup = false;
        try {
            SelectionKey key = this.processKey;
            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
            needWakeup = true;
        } catch (Exception e) {
            AbstractConnection.LOGGER.warn("can't enable write " + e);

        }
        if (needWakeup && wakeup) {
            processKey.selector().wakeup();
        }
    }

注释已经很详细,如此执行完,便成功将握手包发送给了客户端。
在这里稍微吐槽下,由于MyCat在网络通信上同时做了AIO和NIO,但是在设计上AbstractionConnection和这些并没有关系。但是又涉及到缓存队列,所以设计上出现了一些如下的类模式:
这里写图片描述
这样应该是不推荐这么设计的,目前我还没想好如何去改善

© 著作权归作者所有

张哈希_
粉丝 7
博文 42
码字总数 67907
作品 0
房山
程序员
私信 提问
MyCat 看了这篇什么都会了

1、什么是MyCat 1、一个彻底开源的,面向企业应用开发的大数据库集群 2、支持事务、ACID、可以替代MySQL的加强版数据库 3、一个可以视为MySQL集群的企业级数据库,用来替代昂贵的Oracle集群 ...

木迪_2a4e
2018/09/17
0
0
高性能Mysql中间组件Mycat安装配置

高性能Mysql中间组件Mycat安装配置 由于公司项目需要使用到读写分离,但是官方版本的Mysql-proxy读写分离组件太不稳定且有很多BUG,最终决定使用Mycat作为中间组件,其实读写分离只是Mycat中...

幸运券发放
2018/05/02
0
0
MyCAT全局序列号-数据库方式

1.MyCat中的全局序列号介绍 在实现分库分表的情况下,数据库自增主键已无法保证自增主键的全局唯一。为此,MyCat 提供了全局 sequence,并且提供了包含本地配置和数据库配置等多种实现方式。...

HilaryHe
2018/07/23
0
0
Mycat多租户/分布数据库/集群数据库以及SAAS研究进度报告

1、mycat操作 (1)垂直切分配置 (2)水平切分配置 (3)mycat修改schema.xml之后热重启 (4)mycat注解的实现 (5)mycat高可用(搭配HAproxy等一起做集群,多个mycat节点) 2、mycat可以配...

蓝奔z
2018/08/24
0
0
Mycat - 高可用与负载均衡实现,满满的干货!

前情回顾 Mycat - 实现数据库的读写分离与高可用中我们实现了mysql的读写分离与高可用,有几个点我们回顾下 如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring,MyBatis,Net...

编程SHA
02/20
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Jenkins系列_插件安装及报错处理

进入Jenkins之后我们可以进行插件的安装,插件管理位于以下模块: 发现上面报了一堆错误,是因为插件的依赖没有安装好,那么这一节,就先把这些错误解决掉吧。解决完成后,也就基本会使用插件...

shzwork
今天
2
0
mysql mysql的所有查询语句和聚合函数(整理一下,忘记了可以随时看看)

查询所有字段 select * from 表名; 查询自定字段 select 字段名 from 表名; 查询指定数据 select * from 表名 where 条件; 带关键字IN的查询 select * from 表名 where 条件 [not] in(元素...

edison_kwok
昨天
9
0
解决多线程并行加载缓存问题(利用guava实现)

依赖 com.google.guava:guava:20.0 import com.google.common.cache.Cache;import com.google.common.cache.CacheBuilder;import java.util.concurrent.ExecutionException;import j......

暗中观察
昨天
3
0
利用VisualVM 内存查看

准备工作,建几个测试类。等下就是要查看这几个类里面的属性 package visualvm;public class MultiObject { private String str; private int i; MultiObject(String str...

冷基
昨天
2
0
组装一台工作游戏两用机

一、配置清单如下: 分类 项目 价格(元) 主板 华硕(ASUS)TUF Z370-PLUS GAMING II 电竞特工 Z370二代 支持9代CPU 1049 CPU 英特尔(Intel) i7 8700K 酷睿六核 盒装CPU处理器 2640 风扇 九...

mbzhong
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部