文档章节

数据库中间件 MyCAT 源码分析 —— SQL ON MongoDB

芋道源码
 芋道源码
发布于 2017/07/20 01:32
字数 1613
阅读 1921
收藏 81
点赞 2
评论 18

wechat_mp

🙂🙂🙂关注微信公众号有福利:

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言将得到认真回复。甚至不知道如何读源码也可以请教噢
  4. 新的源码解析文章实时收到通知。每周更新一篇左右


1. 概述

可能你在看到这个标题会小小的吃惊,MyCAT 能使用 MongoDB 做数据节点。是的,没错,确实可以。
吼吼吼,让我们开启这段神奇的“旅途”。

本文主要分成四部分:

  1. 总体流程,让你有个整体的认识
  2. 查询操作
  3. 插入操作
  4. 彩蛋,😈彩蛋,🙂彩蛋

建议你看过这两篇文章(非必须):

  1. 《MyCAT 源码分析 —— 【单库单表】插入》
  2. 《MyCAT 源码分析 —— 【单库单表】查询》

2. 主流程

  1. MyCAT Server 接收 MySQL Client 基于 MySQL协议 的请求,翻译 SQLMongoDB操作 发送给 MongoDB Server
  2. MyCAT Server 接收 MongoDB Server 返回的 MongoDB数据,翻译成 MySQL数据结果 返回给 MySQL Client

这样一看,MyCAT 连接 MongoDB 是不是少神奇一点列。


Java数据库连接,(Java Database Connectivity,简称JDBC)是Java语言中用来规范客户端程序如何来访问数据库的应用程序接口,提供了诸如查询和更新数据库中数据的方法。JDBC也是Sun Microsystems的商标。JDBC是面向关系型数据库的。

MyCAT 使用 JDBC 规范,抽象了对 MongoDB 的访问。通过这样的方式,MyCAT 也抽象了 SequoiaDB 的访问。可能这样说法有些抽象,看个类图压压惊。

是不是熟悉的味道。不得不说 JDBC 规范的精妙。

3. 查询操作

SELECT id, name FROM user WHERE name > '' ORDER BY _id DESC;

看顺序图已经很方便的理解整体逻辑,我就不多废话啦。我们来看几个核心的代码逻辑。

1、查询 MongoDB

// MongoSQLParser.java
public MongoData query() throws MongoSQLException {
   if (!(statement instanceof SQLSelectStatement)) {
       //return null;
       throw new IllegalArgumentException("not a query sql statement");
   }
   MongoData mongo = new MongoData();
   DBCursor c = null;
   SQLSelectStatement selectStmt = (SQLSelectStatement) statement;
   SQLSelectQuery sqlSelectQuery = selectStmt.getSelect().getQuery();
   int icount = 0;
   if (sqlSelectQuery instanceof MySqlSelectQueryBlock) {
       MySqlSelectQueryBlock mysqlSelectQuery = (MySqlSelectQueryBlock) selectStmt.getSelect().getQuery();

       BasicDBObject fields = new BasicDBObject();

       // 显示(返回)的字段
       for (SQLSelectItem item : mysqlSelectQuery.getSelectList()) {
           //System.out.println(item.toString());
           if (!(item.getExpr() instanceof SQLAllColumnExpr)) {
               if (item.getExpr() instanceof SQLAggregateExpr) {
                   SQLAggregateExpr expr = (SQLAggregateExpr) item.getExpr();
                   if (expr.getMethodName().equals("COUNT")) { // TODO 待读:count(*)
                       icount = 1;
                       mongo.setField(getExprFieldName(expr), Types.BIGINT);
                   }
                   fields.put(getExprFieldName(expr), 1);
               } else {
                   fields.put(getFieldName(item), 1);
               }
           }

       }

       // 表名
       SQLTableSource table = mysqlSelectQuery.getFrom();
       DBCollection coll = this._db.getCollection(table.toString());
       mongo.setTable(table.toString());

       // WHERE
       SQLExpr expr = mysqlSelectQuery.getWhere();
       DBObject query = parserWhere(expr);

       // GROUP BY
       SQLSelectGroupByClause groupby = mysqlSelectQuery.getGroupBy();
       BasicDBObject gbkey = new BasicDBObject();
       if (groupby != null) {
           for (SQLExpr gbexpr : groupby.getItems()) {
               if (gbexpr instanceof SQLIdentifierExpr) {
                   String name = ((SQLIdentifierExpr) gbexpr).getName();
                   gbkey.put(name, Integer.valueOf(1));
               }
           }
           icount = 2;
       }

       // SKIP / LIMIT
       int limitoff = 0;
       int limitnum = 0;
       if (mysqlSelectQuery.getLimit() != null) {
           limitoff = getSQLExprToInt(mysqlSelectQuery.getLimit().getOffset());
           limitnum = getSQLExprToInt(mysqlSelectQuery.getLimit().getRowCount());
       }
       if (icount == 1) { // COUNT(*)
           mongo.setCount(coll.count(query));
       } else if (icount == 2) { // MapReduce
           BasicDBObject initial = new BasicDBObject();
           initial.put("num", 0);
           String reduce = "function (obj, prev) { " + "  prev.num++}";
           mongo.setGrouyBy(coll.group(gbkey, query, initial, reduce));
       } else {
           if ((limitoff > 0) || (limitnum > 0)) {
               c = coll.find(query, fields).skip(limitoff).limit(limitnum);
           } else {
               c = coll.find(query, fields);
           }

           // order by
           SQLOrderBy orderby = mysqlSelectQuery.getOrderBy();
           if (orderby != null) {
               BasicDBObject order = new BasicDBObject();
               for (int i = 0; i < orderby.getItems().size(); i++) {
                   SQLSelectOrderByItem orderitem = orderby.getItems().get(i);
                   order.put(orderitem.getExpr().toString(), getSQLExprToAsc(orderitem.getType()));
               }
               c.sort(order);
               // System.out.println(order);
           }
       }
       mongo.setCursor(c);
   }
   return mongo;
}

2、查询条件

// MongoSQLParser.java
private void parserWhere(SQLExpr aexpr, BasicDBObject o) {
   if (aexpr instanceof SQLBinaryOpExpr) {
       SQLBinaryOpExpr expr = (SQLBinaryOpExpr) aexpr;
       SQLExpr exprL = expr.getLeft();
       if (!(exprL instanceof SQLBinaryOpExpr)) {
           if (expr.getOperator().getName().equals("=")) {
               o.put(exprL.toString(), getExpValue(expr.getRight()));
           } else {
               String op = "";
               if (expr.getOperator().getName().equals("<")) {
                   op = "$lt";
               } else if (expr.getOperator().getName().equals("<=")) {
                   op = "$lte";
               } else if (expr.getOperator().getName().equals(">")) {
                   op = "$gt";
               } else if (expr.getOperator().getName().equals(">=")) {
                   op = "$gte";
               } else if (expr.getOperator().getName().equals("!=")) {
                   op = "$ne";
               } else if (expr.getOperator().getName().equals("<>")) {
                   op = "$ne";
               }
               parserDBObject(o, exprL.toString(), op, getExpValue(expr.getRight()));
           }
       } else {
           if (expr.getOperator().getName().equals("AND")) {
               parserWhere(exprL, o);
               parserWhere(expr.getRight(), o);
           } else if (expr.getOperator().getName().equals("OR")) {
               orWhere(exprL, expr.getRight(), o);
           } else {
               throw new RuntimeException("Can't identify the operation of  of where");
           }
       }
   }
}

private void orWhere(SQLExpr exprL, SQLExpr exprR, BasicDBObject ob) {
   BasicDBObject xo = new BasicDBObject();
   BasicDBObject yo = new BasicDBObject();
   parserWhere(exprL, xo);
   parserWhere(exprR, yo);
   ob.put("$or", new Object[]{xo, yo});
}

3、解析 MongoDB 数据

// MongoResultSet.java
public MongoResultSet(MongoData mongo, String schema) throws SQLException {
   this._cursor = mongo.getCursor();
   this._schema = schema;
   this._table = mongo.getTable();
   this.isSum = mongo.getCount() > 0;
   this._sum = mongo.getCount();
   this.isGroupBy = mongo.getType();

   if (this.isGroupBy) {
       dblist = mongo.getGrouyBys();
       this.isSum = true;
   }
   if (this._cursor != null) {
       select = _cursor.getKeysWanted().keySet().toArray(new String[0]);
       // 解析 fields
       if (this._cursor.hasNext()) {
           _cur = _cursor.next();
           if (_cur != null) {
               if (select.length == 0) {
                   SetFields(_cur.keySet());
               }
               _row = 1;
           }
       }
       // 设置 fields 类型
       if (select.length == 0) {
           select = new String[]{"_id"};
           SetFieldType(true);
       } else {
           SetFieldType(false);
       }
   } else {
       SetFields(mongo.getFields().keySet());//new String[]{"COUNT(*)"};
       SetFieldType(mongo.getFields());
   }
}
  • 当使用 SELECT * 查询字段时,fields 使用第一条数据返回的 fields。即使,后面的数据有其他 fields,也不返回。

4、返回数据给 MySQL Client

// JDBCConnection.java
private void ouputResultSet(ServerConnection sc, String sql)
       throws SQLException {
   ResultSet rs = null;
   Statement stmt = null;

   try {
       stmt = con.createStatement();
       rs = stmt.executeQuery(sql);

       // header
       List<FieldPacket> fieldPks = new LinkedList<>();
       ResultSetUtil.resultSetToFieldPacket(sc.getCharset(), fieldPks, rs, this.isSpark);
       int colunmCount = fieldPks.size();
       ByteBuffer byteBuf = sc.allocate();
       ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();
       headerPkg.fieldCount = fieldPks.size();
       headerPkg.packetId = ++packetId;
       byteBuf = headerPkg.write(byteBuf, sc, true);
       byteBuf.flip();
       byte[] header = new byte[byteBuf.limit()];
       byteBuf.get(header);
       byteBuf.clear();
       List<byte[]> fields = new ArrayList<byte[]>(fieldPks.size());
       for (FieldPacket curField : fieldPks) {
           curField.packetId = ++packetId;
           byteBuf = curField.write(byteBuf, sc, false);
           byteBuf.flip();
           byte[] field = new byte[byteBuf.limit()];
           byteBuf.get(field);
           byteBuf.clear();
           fields.add(field);
       }
       // header eof
       EOFPacket eofPckg = new EOFPacket();
       eofPckg.packetId = ++packetId;
       byteBuf = eofPckg.write(byteBuf, sc, false);
       byteBuf.flip();
       byte[] eof = new byte[byteBuf.limit()];
       byteBuf.get(eof);
       byteBuf.clear();
       this.respHandler.fieldEofResponse(header, fields, eof, this);

       // row
       while (rs.next()) {
           RowDataPacket curRow = new RowDataPacket(colunmCount);
           for (int i = 0; i < colunmCount; i++) {
               int j = i + 1;
               if (MysqlDefs.isBianry((byte) fieldPks.get(i).type)) {
                   curRow.add(rs.getBytes(j));
               } else if (fieldPks.get(i).type == MysqlDefs.FIELD_TYPE_DECIMAL ||
                       fieldPks.get(i).type == (MysqlDefs.FIELD_TYPE_NEW_DECIMAL - 256)) { // field type is unsigned byte
                   // ensure that do not use scientific notation format
                   BigDecimal val = rs.getBigDecimal(j);
                   curRow.add(StringUtil.encode(val != null ? val.toPlainString() : null, sc.getCharset()));
               } else {
                   curRow.add(StringUtil.encode(rs.getString(j), sc.getCharset()));
               }
           }
           curRow.packetId = ++packetId;
           byteBuf = curRow.write(byteBuf, sc, false);
           byteBuf.flip();
           byte[] row = new byte[byteBuf.limit()];
           byteBuf.get(row);
           byteBuf.clear();
           this.respHandler.rowResponse(row, this);
       }
       fieldPks.clear();
       // row eof
       eofPckg = new EOFPacket();
       eofPckg.packetId = ++packetId;
       byteBuf = eofPckg.write(byteBuf, sc, false);
       byteBuf.flip();
       eof = new byte[byteBuf.limit()];
       byteBuf.get(eof);
       sc.recycle(byteBuf);
       this.respHandler.rowEofResponse(eof, this);
   } finally {
       if (rs != null) {
           try {
               rs.close();
           } catch (SQLException e) {
           }
       }
       if (stmt != null) {
           try {
               stmt.close();
           } catch (SQLException e) {
           }
       }
   }
}

// MongoResultSet.java
@Override
public String getString(String columnLabel) throws SQLException {
   Object x = getObject(columnLabel);
   if (x == null) {
       return null;
   }
   return x.toString();
}
  • 当返回字段值是 Object 时,返回该对象.toString()。例如:
mysql> select * from user order by _id asc;
+--------------------------+------+-------------------------------+
| _id                      | name | profile                       |
+--------------------------+------+-------------------------------+
| 1                        | 123  | { "age" : 1 , "height" : 100} |

4. 插入操作

// MongoSQLParser.java
public int executeUpdate() throws MongoSQLException {
   if (statement instanceof SQLInsertStatement) {
       return InsertData((SQLInsertStatement) statement);
   }
   if (statement instanceof SQLUpdateStatement) {
       return UpData((SQLUpdateStatement) statement);
   }
   if (statement instanceof SQLDropTableStatement) {
       return dropTable((SQLDropTableStatement) statement);
   }
   if (statement instanceof SQLDeleteStatement) {
       return DeleteDate((SQLDeleteStatement) statement);
   }
   if (statement instanceof SQLCreateTableStatement) {
       return 1;
   }
   return 1;
}

private int InsertData(SQLInsertStatement state) {
   if (state.getValues().getValues().size() == 0) {
       throw new RuntimeException("number of  columns error");
   }
   if (state.getValues().getValues().size() != state.getColumns().size()) {
       throw new RuntimeException("number of values and columns have to match");
   }
   SQLTableSource table = state.getTableSource();
   BasicDBObject o = new BasicDBObject();
   int i = 0;
   for (SQLExpr col : state.getColumns()) {
       o.put(getFieldName2(col), getExpValue(state.getValues().getValues().get(i)));
       i++;
   }
   DBCollection coll = this._db.getCollection(table.toString());
   coll.insert(o);
   return 1;
}

5. 彩蛋

老铁,看到这里,来一波微信公众号关注吧?!

wechat_mp

1、支持多 MongoDB ,并使用 MyCAT 进行分片。

MyCAT 配置:multi_mongodb

2、支持 MongoDB + MySQL 作为同一个 MyCAT Table 的数据节点。查询时,可以合并数据结果。

查询时,返回 MySQL 数据记录字段要比 MongoDB 数据记录字段全,否则,合并结果时会报错。

MyCAT 配置:single_mongodb_mysql

3、MongoDB 作为数据节点时,可以使用 MyCAT 提供的数据库主键字段功能。

MyCAT 配置:single_mongodb

© 著作权归作者所有

共有 人打赏支持
芋道源码

芋道源码

粉丝 249
博文 71
码字总数 191189
作品 0
徐汇
后端工程师
加载中

评论(18)

奔跑的potato
奔跑的potato
我还是觉得node.js好用
fengyang
fengyang
不知道放到生产环境上如何
fengyang
fengyang
mycat对mongodb的支持有点弱吧,简单的语法没问题
芋道源码
芋道源码

引用来自“doge_刀戈”的评论

cool
刀哥好,求带 LOL 倔强青铜王者
doge_刀戈
doge_刀戈
cool
芋道源码
芋道源码

引用来自“ljxun”的评论

这种中间件的设计思想有问题,这种中间件应该是放在jdbc之前,通过配置去判断是使用jdbc或是其他,是使用sql语句,还是使用nosql语句。这样才可以模糊数据库的界限,通过解析sql语句去转化成其他语句,这种实现方式太复杂,要做到很全的话,要去重新实现一个sql语句解析器。
看设计的思路,MyCAT 本身的思路目前是“伪装”成 MySQL-Proxy
在 Solr、ES 这些中间件没出来的时候,有些公司是实现 SQL 来去访问 Lucene

MyCAT 目前基于 Druid 实现的语法解析器。

最近在学习 Sharding-JDBC 的源码,之后会出一篇 TA 的语法解析器实现的分享文章。

可以关注我的微信公众号哟。:bowtie:
芋道源码
芋道源码

引用来自“黎明伟”的评论

能够支持的SQL操作应该很有限,是吧
对的。基本是平行翻译的,MongoDB 有的操作,才能被翻译成 SQL
ljxun
ljxun
这种中间件的设计思想有问题,这种中间件应该是放在jdbc之前,通过配置去判断是使用jdbc或是其他,是使用sql语句,还是使用nosql语句。这样才可以模糊数据库的界限,通过解析sql语句去转化成其他语句,这种实现方式太复杂,要做到很全的话,要去重新实现一个sql语句解析器。
黎明伟
黎明伟
能够支持的SQL操作应该很有限,是吧
丁富贵
这东西是个好东西,可惜领头人掉钱眼里了,不知道还有没有心思在开源上,每天忙着卖书卖课程吹牛B,还觉得有竞争对手要谋害自己
MyCAT SQL ON MongoDB

概述 可能你在看到这个标题会小小的吃惊,MyCAT 能使用 MongoDB 做数据节点。是的,没错,确实可以。 吼吼吼,让我们开启这段神奇的“旅途”。 本文主要分成四部分: 总体流程,让你有个整体...

wangchen1999 ⋅ 05/02 ⋅ 0

mongodb通过sql来查询

mongodb本身是不支持用sql进行查询。这是mongodb是nosql数据库。但是可以通过语法转换的方式来实现。记得mycat是有个黑科技的,就是可以查询mongodb。我就在git上查询了一下mycat的源代码。发...

miaojiangmin ⋅ 05/23 ⋅ 0

mycat实现mysql分库分表

1. mycat介绍 MyCat发展到目前的版本,已经不是一个单纯的MySQL代理了,它的后端可以支持MySQL、SQL Server、Oracle、DB2、PostgreSQL等主流数据库,也支持MongoDB这种新型NoSQL方式的存储,...

红隐 ⋅ 05/06 ⋅ 0

myCat schema.xml详解

schema.xml 是mycat重要的配置文件之一,管理者mycat的逻辑库、表、分片规则、DataNode以及DataSource。想要了解使用mycat就要弄清楚schema里面的标签含义,下面将介绍一下schema里面的详细内...

WJXing ⋅ 05/22 ⋅ 0

数据库中间件01-认识mycat

简述 Mycat是国产的一套免费开源的分布式数据库中间件。想必做开发或者运维的朋友对中间件这个词应该是比较熟悉了,我们见到过java中间件,消息中间件等等,这里又来了一个数据库中间件。那么...

细节探索者 ⋅ 05/11 ⋅ 0

Mycat读写分离笔记Windows

Mycat读写分离笔记Windows 自己搭了一个运用Mycat中间件搭建了一个读写分离的demo,昨晚还在奋战当中,因为连接mycat的时候老是报错:No Mycat DataBases selected.尝试了很多的方法,然后还...

推荐码发放 ⋅ 04/29 ⋅ 0

高性能Mysql中间组件Mycat安装配置

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

幸运券发放 ⋅ 05/02 ⋅ 0

MySQL多数据源笔记4-Mycat中间件实战

Mycat 是数据库中间件,就是介于数据库与应用之间,进行数据处理与交互的中间服 务。由于前面讲的对数据进行分片处理之后,从原有的一个库,被切分为多个分片数据库,所有的分片数据库集 群构...

狂小白 ⋅ 03/18 ⋅ 0

Mycat读写分离以及拆库拆表综合实验2:部署配置mycat读写分离与拆库拆表

部署 mycat mycat基于java运行,检查java版本 配置java变量 mycat下载地址 大致架构图 核心概念 schema:由多个数据库组成的逻辑数据库 datanode:将数据切分后,一个大表被分配到不同的分片数...

Mathology ⋅ 05/24 ⋅ 0

使用mycat实现mysql读写分离

使用mycat实现mysql读写分离 mycat在数据层的作用和nginx在应用层很像,都是代理后面实际的服务。mycat实现了mysql协议,对外看起来就是一个mysql数据库,可以和使用普通mysql一样进行操作 ...

肥肥小浣熊 ⋅ 05/06 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

CENTOS7防火墙命令记录

安装Firewall命令: yum install firewalld firewalld-config Firewall开启常见端口命令: firewall-cmd --zone=public --add-port=80/tcp --permanent firewall-cmd --zone=public --add-po......

cavion ⋅ 今天 ⋅ 0

【C++】【STL】利用chromo来测量程序运行时间与日志时间打印精确到微秒

直接上代码吧,没啥好说的。头疼。 #include <iostream>#include <string>#include <ctime>#include <sstream>#include <iomanip>#include <thread>#include <chrono>using ......

muqiusangyang ⋅ 今天 ⋅ 0

Mac环境下svn的使用

在Windows环境中,我们一般使用TortoiseSVN来搭建svn环境。在Mac环境下,由于Mac自带了svn的服务器端和客户端功能,所以我们可以在不装任何第三方软件的前提下使用svn功能,不过还需做一下简...

故久呵呵 ⋅ 今天 ⋅ 0

破解公司回应苹果“USB限制模式”:已攻破

本周四,苹果发表声明称 iOS 中加入了一项名为“USB 限制模式”的功能,可以防止 iPhone 在连接其他设备的时候被破解,并且强调这一功能并不是针对 FBI 等执法部门,为的是保护用户数据安全。...

六库科技 ⋅ 今天 ⋅ 0

MyBtais整合Spring Boot整合,TypeHandler对枚举类(enum)处理

概要 问题描述 我想用枚举类来表示用户当前状态,枚举类由 code 和 msg 组成,但我只想把 code 保存到数据库,查询处理,能知道用户当前状态,这应该怎么做呢?在 Spring 整合MyBatis 的时候...

Wenyi_Feng ⋅ 今天 ⋅ 0

synchronized与Lock的区别

# <center>王梦龙的读书笔记第一篇</center> ## <center>-synchronized与Lock的区别</centre> ###一、从使用场景来说 + synchronized 是能够注释代码块、类、方法但是它的加锁是和解锁使用一......

我不想加班 ⋅ 今天 ⋅ 0

VConsole的使用

手机端控制台打印输出,方便bug的排查。 首先需要引入vconsole.min.js 文件,然后在文件中创造实例。就能直接使用了。 var vConsole = new VConsole(); vConsole的文件地址...

大美琴 ⋅ 今天 ⋅ 0

Java NIO之字符集

1 字符集和编解码的概念 首先,解释一下什么是字符集。顾名思义,就是字符的集合。它的初衷是把现实世界的符号映射为计算机可以理解的字节。比如我创造一个字符集,叫做sex字符集,就包含两个...

士别三日 ⋅ 今天 ⋅ 0

Spring Bean基础

1、Bean之间引用 <!--如果Bean配置在同一个XML文件中,使用local引用--><ref bean="someBean"/><!--如果Bean配置在不同的XML文件中,使用ref引用--><ref local="someBean"/> 其实两种......

霍淇滨 ⋅ 今天 ⋅ 0

05、基于Consul+Upsync+Nginx实现动态负载均衡

1、Consul环境搭建 下载consul_0.7.5_linux_amd64.zip到/usr/local/src目录 cd /usr/local/srcwget https://releases.hashicorp.com/consul/0.7.5/consul_0.7.5_linux_amd64.zip 解压consu......

北岩 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部