如何進行數據庫中間件 MyCAT 源碼分析

這篇文章將為大家詳細講解有關如何進行數據庫中間件 MyCAT 源碼分析，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

1. 概述

可能你在看到這個標題會小小的吃驚，MyCAT 能使用 MongoDB 做數據節點。是的，沒錯，確實可以。

吼吼吼，讓我們開啟這段神奇的“旅途”。

本文主要分成四部分：

1. 總體流程，讓你有個整體的認識

2. 查詢操作

3. 插入操作

2. 主流程

MyCAT Server 接收 MySQL Client 基于 MySQL協議 的請求，翻譯 SQL 成 MongoDB操作 發送給 MongoDB  Server。

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

關于如何進行數據庫中間件 MyCAT 源碼分析就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。