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

1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

4. 新的源码解析文章实时收到通知。每周更新一篇左右。

• 1. 概述

• 2. 主流程

• 3. 查询操作

• 4. 插入操作

• 5. 彩蛋

1. 概述

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

本文主要分成四部分：

1. 总体流程，让你有个整体的认识

2. 查询操作

3. 插入操作

4. 彩蛋，😈彩蛋，🙂彩蛋

建议你看过这两篇文章（非必须）：

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

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

2. 主流程

1. MyCATServer 接收 MySQLClient 基于 MySQL协议 的请求，翻译 SQL 成 MongoDB操作 发送给 MongoDBServer。

2. MyCATServer 接收 MongoDBServer 返回的 MongoDB数据，翻译成 MySQL数据结果 返回给 MySQLClient。

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

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

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

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

3. 查询操作

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

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

1、查询 MongoDB

1. // MongoSQLParser.java

2. public MongoData query() throws MongoSQLException {

3.   if (!(statement instanceof SQLSelectStatement)) {

4.       //return null;

5.       throw new IllegalArgumentException("not a query sql statement");

6.   }

7.   MongoData mongo = new MongoData();

8.   DBCursor c = null;

9.   SQLSelectStatement selectStmt = (SQLSelectStatement) statement;

10.   SQLSelectQuery sqlSelectQuery = selectStmt.getSelect().getQuery();

11.   int icount = 0;

12.   if (sqlSelectQuery instanceof MySqlSelectQueryBlock) {

13.       MySqlSelectQueryBlock mysqlSelectQuery = (MySqlSelectQueryBlock) selectStmt.getSelect().getQuery();

15.       BasicDBObject fields = new BasicDBObject();

17.       // 显示（返回）的字段

18.       for (SQLSelectItem item : mysqlSelectQuery.getSelectList()) {

19.           //System.out.println(item.toString());

20.           if (!(item.getExpr() instanceof SQLAllColumnExpr)) {

21.               if (item.getExpr() instanceof SQLAggregateExpr) {

22.                   SQLAggregateExpr expr = (SQLAggregateExpr) item.getExpr();

23.                   if (expr.getMethodName().equals("COUNT")) { // TODO 待读：count（*）

24.                       icount = 1;

25.                       mongo.setField(getExprFieldName(expr), Types.BIGINT);

26.                   }

27.                   fields.put(getExprFieldName(expr), 1);

28.               } else {

29.                   fields.put(getFieldName(item), 1);

30.               }

31.           }

33.       }

35.       // 表名

36.       SQLTableSource table = mysqlSelectQuery.getFrom();

37.       DBCollection coll = this._db.getCollection(table.toString());

38.       mongo.setTable(table.toString());

40.       // WHERE

41.       SQLExpr expr = mysqlSelectQuery.getWhere();

42.       DBObject query = parserWhere(expr);

44.       // GROUP BY

45.       SQLSelectGroupByClause groupby = mysqlSelectQuery.getGroupBy();

46.       BasicDBObject gbkey = new BasicDBObject();

47.       if (groupby != null) {

48.           for (SQLExpr gbexpr : groupby.getItems()) {

49.               if (gbexpr instanceof SQLIdentifierExpr) {

50.                   String name = ((SQLIdentifierExpr) gbexpr).getName();

51.                   gbkey.put(name, Integer.valueOf(1));

52.               }

53.           }

54.           icount = 2;

55.       }

57.       // SKIP / LIMIT

58.       int limitoff = 0;

59.       int limitnum = 0;

60.       if (mysqlSelectQuery.getLimit() != null) {

61.           limitoff = getSQLExprToInt(mysqlSelectQuery.getLimit().getOffset());

62.           limitnum = getSQLExprToInt(mysqlSelectQuery.getLimit().getRowCount());

63.       }

64.       if (icount == 1) { // COUNT（*）

65.           mongo.setCount(coll.count(query));

66.       } else if (icount == 2) { // MapReduce

67.           BasicDBObject initial = new BasicDBObject();

68.           initial.put("num", 0);

69.           String reduce = "function (obj, prev) { " + "  prev.num++}";

70.           mongo.setGrouyBy(coll.group(gbkey, query, initial, reduce));

71.       } else {

72.           if ((limitoff > 0) || (limitnum > 0)) {

73.               c = coll.find(query, fields).skip(limitoff).limit(limitnum);

74.           } else {

75.               c = coll.find(query, fields);

76.           }

78.           // order by

79.           SQLOrderBy orderby = mysqlSelectQuery.getOrderBy();

80.           if (orderby != null) {

81.               BasicDBObject order = new BasicDBObject();

82.               for (int i = 0; i < orderby.getItems().size(); i++) {

83.                   SQLSelectOrderByItem orderitem = orderby.getItems().get(i);

84.                   order.put(orderitem.getExpr().toString(), getSQLExprToAsc(orderitem.getType()));

85.               }

86.               c.sort(order);

87.               // System.out.println(order);

88.           }

89.       }

90.       mongo.setCursor(c);

91.   }

92.   return mongo;

93. }

2、查询条件

1. // MongoSQLParser.java

2. private void parserWhere(SQLExpr aexpr, BasicDBObject o) {

3.   if (aexpr instanceof SQLBinaryOpExpr) {

4.       SQLBinaryOpExpr expr = (SQLBinaryOpExpr) aexpr;

5.       SQLExpr exprL = expr.getLeft();

6.       if (!(exprL instanceof SQLBinaryOpExpr)) {

7.           if (expr.getOperator().getName().equals("=")) {

8.               o.put(exprL.toString(), getExpValue(expr.getRight()));

9.           } else {

10.               String op = "";

11.               if (expr.getOperator().getName().equals("<")) {

12.                   op = "$lt";

13.               } else if (expr.getOperator().getName().equals("<=")) {

14.                   op = "$lte";

15.               } else if (expr.getOperator().getName().equals(">")) {

16.                   op = "$gt";

17.               } else if (expr.getOperator().getName().equals(">=")) {

18.                   op = "$gte";

19.               } else if (expr.getOperator().getName().equals("!=")) {

20.                   op = "$ne";

21.               } else if (expr.getOperator().getName().equals("<>")) {

22.                   op = "$ne";

23.               }

24.               parserDBObject(o, exprL.toString(), op, getExpValue(expr.getRight()));

25.           }

26.       } else {

27.           if (expr.getOperator().getName().equals("AND")) {

28.               parserWhere(exprL, o);

29.               parserWhere(expr.getRight(), o);

30.           } else if (expr.getOperator().getName().equals("OR")) {

31.               orWhere(exprL, expr.getRight(), o);

32.           } else {

33.               throw new RuntimeException("Can't identify the operation of  of where");

34.           }

35.       }

36.   }

37. }

39. private void orWhere(SQLExpr exprL, SQLExpr exprR, BasicDBObject ob) {

40.   BasicDBObject xo = new BasicDBObject();

41.   BasicDBObject yo = new BasicDBObject();

42.   parserWhere(exprL, xo);

43.   parserWhere(exprR, yo);

44.   ob.put("$or", new Object[]{xo, yo});

45. }

3、解析 MongoDB 数据

1. // MongoResultSet.java

2. public MongoResultSet(MongoData mongo, String schema) throws SQLException {

3.   this._cursor = mongo.getCursor();

4.   this._schema = schema;

5.   this._table = mongo.getTable();

6.   this.isSum = mongo.getCount() > 0;

7.   this._sum = mongo.getCount();

8.   this.isGroupBy = mongo.getType();

10.   if (this.isGroupBy) {

11.       dblist = mongo.getGrouyBys();

12.       this.isSum = true;

13.   }

14.   if (this._cursor != null) {

15.       select = _cursor.getKeysWanted().keySet().toArray(new String[0]);

16.       // 解析 fields

17.       if (this._cursor.hasNext()) {

18.           _cur = _cursor.next();

19.           if (_cur != null) {

20.               if (select.length == 0) {

21.                   SetFields(_cur.keySet());

22.               }

23.               _row = 1;

24.           }

25.       }

26.       // 设置 fields 类型

27.       if (select.length == 0) {

28.           select = new String[]{"_id"};

29.           SetFieldType(true);

30.       } else {

31.           SetFieldType(false);

32.       }

33.   } else {

34.       SetFields(mongo.getFields().keySet());//new String[]{"COUNT(*)"};

35.       SetFieldType(mongo.getFields());

36.   }

37. }

• 当使用 SELECT* 查询字段时，fields 使用第一条数据返回的 fields。即使，后面的数据有其他 fields，也不返回。

4、返回数据给 MySQL Client

1. // JDBCConnection.java

2. private void ouputResultSet(ServerConnection sc, String sql)

3.       throws SQLException {

4.   ResultSet rs = null;

5.   Statement stmt = null;

7.   try {

8.       stmt = con.createStatement();

9.       rs = stmt.executeQuery(sql);

11.       // header

12.       List<FieldPacket> fieldPks = new LinkedList<>();

13.       ResultSetUtil.resultSetToFieldPacket(sc.getCharset(), fieldPks, rs, this.isSpark);

14.       int colunmCount = fieldPks.size();

15.       ByteBuffer byteBuf = sc.allocate();

16.       ResultSetHeaderPacket headerPkg = new ResultSetHeaderPacket();

17.       headerPkg.fieldCount = fieldPks.size();

18.       headerPkg.packetId = ++packetId;

19.       byteBuf = headerPkg.write(byteBuf, sc, true);

20.       byteBuf.flip();

21.       byte[] header = new byte[byteBuf.limit()];

22.       byteBuf.get(header);

23.       byteBuf.clear();

24.       List<byte[]> fields = new ArrayList<byte[]>(fieldPks.size());

25.       for (FieldPacket curField : fieldPks) {

26.           curField.packetId = ++packetId;

27.           byteBuf = curField.write(byteBuf, sc, false);

28.           byteBuf.flip();

29.           byte[] field = new byte[byteBuf.limit()];

30.           byteBuf.get(field);

31.           byteBuf.clear();

32.           fields.add(field);

33.       }

34.       // header eof

35.       EOFPacket eofPckg = new EOFPacket();

36.       eofPckg.packetId = ++packetId;

37.       byteBuf = eofPckg.write(byteBuf, sc, false);

38.       byteBuf.flip();

39.       byte[] eof = new byte[byteBuf.limit()];

40.       byteBuf.get(eof);

41.       byteBuf.clear();

42.       this.respHandler.fieldEofResponse(header, fields, eof, this);

44.       // row

45.       while (rs.next()) {

46.           RowDataPacket curRow = new RowDataPacket(colunmCount);

47.           for (int i = 0; i < colunmCount; i++) {

48.               int j = i + 1;

49.               if (MysqlDefs.isBianry((byte) fieldPks.get(i).type)) {

50.                   curRow.add(rs.getBytes(j));

51.               } else if (fieldPks.get(i).type == MysqlDefs.FIELD_TYPE_DECIMAL ||

52.                       fieldPks.get(i).type == (MysqlDefs.FIELD_TYPE_NEW_DECIMAL - 256)) { // field type is unsigned byte

53.                   // ensure that do not use scientific notation format

54.                   BigDecimal val = rs.getBigDecimal(j);

55.                   curRow.add(StringUtil.encode(val != null ? val.toPlainString() : null, sc.getCharset()));

56.               } else {

57.                   curRow.add(StringUtil.encode(rs.getString(j), sc.getCharset()));

58.               }

59.           }

60.           curRow.packetId = ++packetId;

61.           byteBuf = curRow.write(byteBuf, sc, false);

62.           byteBuf.flip();

63.           byte[] row = new byte[byteBuf.limit()];

64.           byteBuf.get(row);

65.           byteBuf.clear();

66.           this.respHandler.rowResponse(row, this);

67.       }

68.       fieldPks.clear();

69.       // row eof

70.       eofPckg = new EOFPacket();

71.       eofPckg.packetId = ++packetId;

72.       byteBuf = eofPckg.write(byteBuf, sc, false);

73.       byteBuf.flip();

74.       eof = new byte[byteBuf.limit()];

75.       byteBuf.get(eof);

76.       sc.recycle(byteBuf);

77.       this.respHandler.rowEofResponse(eof, this);

78.   } finally {

79.       if (rs != null) {

80.           try {

81.               rs.close();

82.           } catch (SQLException e) {

83.           }

84.       }

85.       if (stmt != null) {

86.           try {

87.               stmt.close();

88.           } catch (SQLException e) {

89.           }

90.       }

91.   }

92. }

94. // MongoResultSet.java

95. @Override

96. public String getString(String columnLabel) throws SQLException {

97.   Object x = getObject(columnLabel);

98.   if (x == null) {

99.       return null;

100.   }

101.   return x.toString();

102. }

• 当返回字段值是 Object 时，返回该对象.toString()。例如：

1. mysql> select * from user order by _id asc;

2. +--------------------------+------+-------------------------------+

3. | _id                      | name | profile                       |

4. +--------------------------+------+-------------------------------+

5. | 1                        | 123  | { "age" : 1 , "height" : 100} |

4. 插入操作

1. // MongoSQLParser.java

2. public int executeUpdate() throws MongoSQLException {

3.   if (statement instanceof SQLInsertStatement) {

4.       return InsertData((SQLInsertStatement) statement);

5.   }

6.   if (statement instanceof SQLUpdateStatement) {

7.       return UpData((SQLUpdateStatement) statement);

8.   }

9.   if (statement instanceof SQLDropTableStatement) {

10.       return dropTable((SQLDropTableStatement) statement);

11.   }

12.   if (statement instanceof SQLDeleteStatement) {

13.       return DeleteDate((SQLDeleteStatement) statement);

14.   }

15.   if (statement instanceof SQLCreateTableStatement) {

16.       return 1;

17.   }

18.   return 1;

19. }

21. private int InsertData(SQLInsertStatement state) {

22.   if (state.getValues().getValues().size() == 0) {

23.       throw new RuntimeException("number of  columns error");

24.   }

25.   if (state.getValues().getValues().size() != state.getColumns().size()) {

26.       throw new RuntimeException("number of values and columns have to match");

27.   }

28.   SQLTableSource table = state.getTableSource();

29.   BasicDBObject o = new BasicDBObject();

30.   int i = 0;

31.   for (SQLExpr col : state.getColumns()) {

32.       o.put(getFieldName2(col), getExpValue(state.getValues().getValues().get(i)));

33.       i++;

34.   }

35.   DBCollection coll = this._db.getCollection(table.toString());

36.   coll.insert(o);

37.   return 1;

38. }

5. 彩蛋

老铁，看到这里，来一波微信公众号关注吧？！

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

MyCAT 配置：multi_mongodb

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

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

MyCAT 配置：singlemongodbmysql

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

MyCAT 配置：single_mongodb