如何通过binlog获取我们想要的MySql语句?

前言

MySql的binlog一般用于我们对数据的恢复，以及从数据库对主数据库的复制和更新。

假设此时我们有一个需要查询和读取Mysql最近操作DDL的信息，我们需要怎么处理？

聪明的你可能已经想到了，我们可以使用mysqlbinlog工具读取啊！的确，mysqlbinlog对于statement或者mixed格式的binlog文件确实会很方便读取，但是你要知道，从Mysql5.7.7开始，row就是默认的binlog_format，此时我们再要去直接通过肉眼去看，恐怕就不是那么容易了。

即使我们在通过mysqlbinlog解析时加上-v参数，也只能显示出这样的效果:

于是，我写了一个binlog2sql的初级版本，来实现sql语句的转换。

实现

实现过程不是很复杂，主要是通过mysqlbinlog来提取我们需要的DDL语句，然后我们再通过我们的方法来把这些语句转化为我们可以识别的sql语句。

核心代码:

/**
     * @return $this
     */
    protected function selectFromBinLog()
    {
        $fillFile = Util::getFile(__DIR__ . '/data/file.sql');
        file_put_contents($fillFile, "");
        exec("mysqlbinlog -v --database='" . Conf::__DATABASE__ . "' $this->_binlog_basename/$this->_binlog_file | grep -E -i '###|UPDATE|INSERT|DELETE' >> $fillFile");
        return $this;
    }    /**
     * @return $this
     */
    protected function parseSql()
    {
        $fillFileHandler = fopen(__DIR__ . '/data/file.sql', 'r');
        $sqlArr = [];
        if ($this->_type == 'ROW') {
            $match = NULL;
            $sqlStr = "";
            while (($sql = fgets($fillFileHandler)) !== false) {
                if (($match = preg_match('/UPDATE|INSERT|DELETE/', $sql)) || strrpos($sql, 'end_log_pos') !== false) {
                    # 如果有指定表
                    if ($match && $this->_table && strpos($sql, $this->_table) === false) continue;
                    $sqlStr == '' || array_push($sqlArr, $sqlStr);
                    $sqlStr = $match ? trim(substr($sql, 3, -1)) . " " : "";
                } elseif (strpos($sql, '@') !== false || strpos($sql, 'SET')) {
                    $sqlStr .= trim(substr($sql, 3, -1)) . " ";
                }
            }
            $sqlStr == '' || array_push($sqlArr, $sqlStr);
        } else {
            # statement 和 mixed格式一样
            while (($sql = fgets($fillFileHandler)) !== false) {
                $sql = trim($sql);
                if (preg_match('/(UPDATE|INSERT|DELETE)\s+/', $sql)) {
                    array_push($sqlArr, $sql);
                }
            }
        }
        $sqlArr = array_map(function ($value) {
            return preg_replace_callback('/(@(\d+))/', function ($matches) use ($value) {
                $parts = explode('.', $value);
                return $this->getTableColumns(explode('`', array_pop($parts))[1])[$matches[2] - 1];
            }, $value);
        }, $sqlArr);        $mysqlFile = Util::getFile(__DIR__ . '/data/mysql.sql');        array_map(function ($value) use ($mysqlFile) {
            file_put_contents($mysqlFile, $value . PHP_EOL, FILE_APPEND);
        }, $sqlArr);
        fclose($fillFileHandler);        return $this;
    }    /**
     * @param $table
     * @return array
     */
    protected function getTableColumns($table)
    {
        if (array_key_exists($table, $this->_tableColumns))
            return $this->_tableColumns[$table];
        $tableInfo = $this->select("show full columns from $table");
        if (empty($tableInfo)) Util::dd("$table 不存在");
        return $this->_tableColumns[$table] = array_column($tableInfo, 'Field');
    }