src/SMProxyServer.php
private function query(BinaryPacket $bin, string &$data, int $fd)
    {
        $trim_data = rtrim($data);
        switch ($bin->data[4]) {
            case MySQLPacket::$COM_INIT_DB:
                // just init the frontend
                break;
            case MySQLPacket::$COM_STMT_PREPARE:
            case MySQLPacket::$COM_QUERY:
                $connection = new FrontendConnection();
                $queryType = $connection->query($bin);
                $hintArr   = RouteService::route(substr($data, 5, strlen($data) - 5));
                if (isset($hintArr['db_type'])) {
                    switch ($hintArr['db_type']) {
                        case 'read':
                            if ($queryType == ServerParse::DELETE || $queryType == ServerParse::INSERT ||
                                $queryType == ServerParse::REPLACE || $queryType == ServerParse::UPDATE ||
                                $queryType == ServerParse::DDL) {
                                $this->connectReadState[$fd] = false;
                                $system_log = Log::getLogger('system');
                                $system_log->warning("should not use hint 'db_type' to route 'delete', 'insert', 'replace', 'update', 'ddl' to a slave db.");
                            } else {
                                $this->connectReadState[$fd] = true;
                            }
                            break;
                        case 'write':
                            $this->connectReadState[$fd] = false;
                            break;
                        default:
                            $this->connectReadState[$fd] = false;
                            $system_log = Log::getLogger('system');
                            $system_log->warning("use hint 'db_type' value is not found.");
                            break;
                    }
                } elseif (ServerParse::SELECT == $queryType ||
                    ServerParse::SHOW == $queryType ||
                    (ServerParse::SET == $queryType && false === strpos($data, 'autocommit', 4)) ||
                    ServerParse::USE == $queryType
                ) {
                    //处理读操作
                    if (!isset($this->connectHasTransaction[$fd]) ||
                        !$this->connectHasTransaction[$fd]) {
                        if ((('u' == $trim_data[-6] || 'U' == $trim_data[-6]) &&
                            ServerParse::UPDATE == ServerParse::uCheck($trim_data, -6, false))) {
                            //判断悲观锁
                            $this->connectReadState[$fd] = false;
                        } else {
                            $this->connectReadState[$fd] = true;
                        }
                    }
                } elseif (ServerParse::START == $queryType || ServerParse::BEGIN == $queryType
                ) {
                    //处理事务
                    $this->connectHasTransaction[$fd] = true;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::SET == $queryType && false !== strpos($data, 'autocommit', 4) &&
                    0 == $trim_data[-1]) {
                    //处理autocommit事务
                    $this->connectHasAutoCommit[$fd] = true;
                    $this->connectHasTransaction[$fd] = true;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::SET == $queryType && false !== strpos($data, 'autocommit', 4) &&
                    1 == $trim_data[-1]) {
                    $this->connectHasAutoCommit[$fd] = false;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::COMMIT == $queryType || ServerParse::ROLLBACK == $queryType) {
                    //事务提交
                    $this->connectHasTransaction[$fd] = false;
                } else {
                    $this->connectReadState[$fd] = false;
                }
                break;
            case MySQLPacket::$COM_PING:
                break;
            case MySQLPacket::$COM_QUIT:
                //禁用客户端退出
                $data = '';
                break;
            case MySQLPacket::$COM_PROCESS_KILL:
                break;
            case MySQLPacket::$COM_STMT_EXECUTE:
                break;
            case MySQLPacket::$COM_STMT_CLOSE:
                break;
            case MySQLPacket::$COM_HEARTBEAT:
                break;
            default:
                break;
        }
    }
private function query(BinaryPacket $bin, string &$data, int $fd)
    {
        $trim_data = rtrim($data);
        switch ($bin->data[4]) {
            case MySQLPacket::$COM_INIT_DB:
                // just init the frontend
                break;
            case MySQLPacket::$COM_STMT_PREPARE:
            case MySQLPacket::$COM_QUERY:
                $connection = new FrontendConnection();
                $queryType = $connection->query($bin);
                $hintArr   = RouteService::route(substr($data, 5, strlen($data) - 5));
                if (isset($hintArr['db_type'])) {
                    switch ($hintArr['db_type']) {
                        case 'read':
                            if ($queryType == ServerParse::DELETE || $queryType == ServerParse::INSERT ||
                                $queryType == ServerParse::REPLACE || $queryType == ServerParse::UPDATE ||
                                $queryType == ServerParse::DDL) {
                                $this->connectReadState[$fd] = false;
                                $system_log = Log::getLogger('system');
                                $system_log->warning("should not use hint 'db_type' to route 'delete', 'insert', 'replace', 'update', 'ddl' to a slave db.");
                            } else {
                                $this->connectReadState[$fd] = true;
                            }
                            break;
                        case 'write':
                            $this->connectReadState[$fd] = false;
                            break;
                        default:
                            $this->connectReadState[$fd] = false;
                            $system_log = Log::getLogger('system');
                            $system_log->warning("use hint 'db_type' value is not found.");
                            break;
                    }
                } elseif (ServerParse::SELECT == $queryType ||
                    ServerParse::SHOW == $queryType ||
                    (ServerParse::SET == $queryType && false === strpos($data, 'autocommit', 4)) ||
                    ServerParse::USE == $queryType
                ) {
                    //处理读操作
                    if (!isset($this->connectHasTransaction[$fd]) ||
                        !$this->connectHasTransaction[$fd]) {
                        if ((('u' == $trim_data[-6] || 'U' == $trim_data[-6]) &&
                            ServerParse::UPDATE == ServerParse::uCheck($trim_data, -6, false))) {
                            //判断悲观锁
                            $this->connectReadState[$fd] = false;
                        } else {
                            $this->connectReadState[$fd] = true;
                        }
                    }
                } elseif (ServerParse::START == $queryType || ServerParse::BEGIN == $queryType
                ) {
                    //处理事务
                    $this->connectHasTransaction[$fd] = true;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::SET == $queryType && false !== strpos($data, 'autocommit', 4) &&
                    0 == $trim_data[-1]) {
                    //处理autocommit事务
                    $this->connectHasAutoCommit[$fd] = true;
                    $this->connectHasTransaction[$fd] = true;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::SET == $queryType && false !== strpos($data, 'autocommit', 4) &&
                    1 == $trim_data[-1]) {
                    $this->connectHasAutoCommit[$fd] = false;
                    $this->connectReadState[$fd] = false;
                } elseif (ServerParse::COMMIT == $queryType || ServerParse::ROLLBACK == $queryType) {
                    //事务提交
                    $this->connectHasTransaction[$fd] = false;
                } else {
                    $this->connectReadState[$fd] = false;
                }
                break;
            case MySQLPacket::$COM_PING:
                break;
            case MySQLPacket::$COM_QUIT:
                //禁用客户端退出
                $data = '';
                break;
            case MySQLPacket::$COM_PROCESS_KILL:
                break;
            case MySQLPacket::$COM_STMT_EXECUTE:
                break;
            case MySQLPacket::$COM_STMT_CLOSE:
                break;
            case MySQLPacket::$COM_HEARTBEAT:
                break;
            default:
                break;
        }
    }
语句解析处理

@param BinaryPacket $bin
@param string $data
@param int $fd

@throws MySQLException