MySql binlog恢復數據

1. 直接導入數據庫

mysqlbinlog --database=testdb mysql-bin.000001 | mysql -uroot -f

2. 導出成SQL文

(1) 從binlog輸出為SQL

mysqlbinlog -vv --database=testdb --base64-output=decode-rows mysql-bin.000001 > 0001.sql
grep "###" 0001.sql >  0001_#.sql

導出的SQL文如下格式：

### UPDATE `test`.`test_data1`
### WHERE
###   @1='2002001809730903086' /* VARSTRING(150) meta=150 nullable=0 is_null=0 */
###   @2='2001001959581442003' /* VARSTRING(150) meta=150 nullable=0 is_null=0 */
###   @3='0661594' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @4='758686' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @5=1449174648065 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @6=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @7=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @8=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @9=1458787027688 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @10=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @11=1458787027688 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @12=300 /* INT meta=0 nullable=1 is_null=0 */
### SET
###   @1='2002001809730903086' /* VARSTRING(150) meta=150 nullable=0 is_null=0 */
###   @2='2001001959581442003' /* VARSTRING(150) meta=150 nullable=0 is_null=0 */
###   @3='0661594' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @4='758686' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @5=1449174648065 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @6=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @7=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @8=0 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @9=1458787027688 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @10=1458787191658 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @11=1458787191658 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @12=400 /* INT meta=0 nullable=1 is_null=0 */
### INSERT INTO `test`.`test_data2`
### SET
###   @1='00020017b4c3418b' /* VARSTRING(48) meta=48 nullable=0 is_null=0 */
###   @2='2002001809730903086' /* VARSTRING(150) meta=150 nullable=0 is_null=0 */
###   @3='00003997' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @4='0661594' /* VARSTRING(150) meta=150 nullable=1 is_null=0 */
###   @5=1458787191658 /* LONGINT meta=0 nullable=1 is_null=0 */
###   @6=0 /* INT meta=0 nullable=1 is_null=0 */

(2) 需要將其轉換可以在mysql客戶端執行的SQL。我寫了一個小程序處理這個格式，代碼如下：

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 *
 * @author Beef Liu
 * 
 **/
public class BinLogSQLConverter {
    public final static String RTN_LINE = "\r\n";
    
    private final static String SQL_WORD_INSERT_INTO = "INSERT INTO";
    private final static String SQL_WORD_UPDATE = "UPDATE";
    private final static String SQL_WORD_DELETE = "DELETE";
    
    private final static String SQL_WORD_SET = "SET";
    private final static String SQL_WORD_WHERE = "WHERE";

    private final static String LINE_COMMENT = "###";
    private final static String REGEX_BLOCK_COMMENT = "/\\*[^\\n']*/";
    
    public static class TableDesc {
        private String _tableName;
        private String[] _colNames;
        
        public TableDesc() {
        }
        
        public TableDesc(String tableName, String[] colNames) {
            _tableName = tableName;
            _colNames = colNames;
        }
        
        public String getTableName() {
            return _tableName;
        }
        public void setTableName(String tableName) {
            _tableName = tableName;
        }
        public String[] getColNames() {
            return _colNames;
        }
        public void setColNames(String[] colNames) {
            _colNames = colNames;
        }
    }
    
    public static void convertToSQL(
            File inputBinSQL, File outputSQL,
            Charset charset,
            TableDesc[] tableDescs
            ) throws IOException {
        BufferedReader reader = new BufferedReader(
                new InputStreamReader(new FileInputStream(inputBinSQL), charset));
        
        try {
            OutputStreamWriter writer = new OutputStreamWriter(
                    new BufferedOutputStream(new FileOutputStream(outputSQL)),
                    charset);
            try {
                List<String> lines = new ArrayList<String>();
                
                while(true) {
                    String line = reader.readLine();
                    if(line == null) {
                        break;
                    }
                    
                    //trim '###'
                    if(line.startsWith(LINE_COMMENT)) {
                        line = line.substring(LINE_COMMENT.length());
                    }
                    
                    //trim block comment
                    line = line.replaceAll(REGEX_BLOCK_COMMENT, "");

                    if(startsWithAndIgnoreSpace(line, SQL_WORD_INSERT_INTO)
                            || startsWithAndIgnoreSpace(line, SQL_WORD_UPDATE)
                            || startsWithAndIgnoreSpace(line, SQL_WORD_DELETE)
                            ) {
                        //format previous statement
                        if(lines.size() > 0) {
                            String formatedStmt = formatStatement(tableDescs, lines);
                            if(formatedStmt != null && formatedStmt.length() > 0) {
                                writer.append(formatedStmt).append(RTN_LINE);
                            }
                        }
                        
                        //new statement
                        lines.clear();
                    }
                    
                    lines.add(line);
                }
                
                if(lines.size() > 0) {
                    String formatedStmt = formatStatement(tableDescs, lines);
                    if(formatedStmt != null && formatedStmt.length() > 0) {
                        writer.append(formatedStmt).append(RTN_LINE);
                    }
                }
            } finally {
                writer.close();
            }
        } finally {
            reader.close();
        }
    }
    
    private final static String formatStatement(TableDesc[] tableDescs, List<String> lines) {
        String line;
        
        line = lines.get(0);
        if(startsWithAndIgnoreSpace(line, SQL_WORD_INSERT_INTO)) {
            return formatStatementOfInsert(tableDescs, lines);
        } else if (startsWithAndIgnoreSpace(line, SQL_WORD_UPDATE)) {
            return formatStatementOfUpdate(tableDescs, lines);
        } else {
            System.out.println("Not supported yet. statement type:" + line);
            return "";
        }
    }
    
    private final static String formatStatementOfInsert(TableDesc[] tableDescs, List<String> lines) {
        /*
 INSERT INTO `test`.`test_data1`
 SET
   @1='00020017bae7830b' 
   @2='2002001280299806243' 
   @3='00004407' 
   @4='0661601' 
   @5=1458889871654 
   @6=0 
         */

        String tableName = findTableNameInLine(lines.get(0));
        TableDesc tblDesc = getTableDesc(tableDescs, tableName);
        if(tblDesc == null) {
            return null;
        }
        
        StringBuilder sql = new StringBuilder();
        //INSERT INTO
        sql.append(lines.get(0)).append(RTN_LINE);
        //SET
        sql.append(lines.get(1)).append(RTN_LINE);
        
        //@1=xxxx
        for(int i = 2; i < lines.size(); i++) {
            if (i > 2) {
                sql.append(", ");
            }
            sql.append(substituteColName(tblDesc, lines.get(i))).append(RTN_LINE);
        }
        
        sql.append(";").append(RTN_LINE);
        
        return sql.toString();
    }
    
    private final static String formatStatementOfUpdate(TableDesc[] tableDescs, List<String> lines) {
        /*
 UPDATE `test`.`test_data2`
 WHERE
   @1='2002001280299806243' 
   @2='2001001394921721184' 
   @3='0661601' 
   @4='758029' 
   @5=1450922125360 
   @6=0 
   @7=1458889670699 
   @8=0 
   @9=1458889682247 
   @10=0 
   @11=1458889682247 
   @12=300 
 SET
   @1='2002001280299806243' 
   @2='2001001394921721184' 
   @3='0661601' 
   @4='758029' 
   @5=1450922125360 
   @6=0 
   @7=1458889670699 
   @8=0 
   @9=1458889682247 
   @10=1458889871654 
   @11=1458889871654 
   @12=400 
         */
        
        String tableName = findTableNameInLine(lines.get(0));
        TableDesc tblDesc = getTableDesc(tableDescs, tableName);
        if(tblDesc == null) {
            return null;
        }
        
        StringBuilder sql = new StringBuilder();
        //UPDATE
        sql.append(lines.get(0)).append(RTN_LINE);
        
        List<String> blockOfWhere = new ArrayList<String>(); 
        List<String> blockOfSet = new ArrayList<String>();
        
        //@1=xxxx
        List<String> blockRef = blockOfWhere;
        for(int i = 1; i < lines.size(); i++) {
            String line = lines.get(i);
            
            if(startsWithAndIgnoreSpace(line, SQL_WORD_SET)) {
                blockRef = blockOfSet;
            }
            
            line = substituteColName(tblDesc, line);
            blockRef.add(line);
        }
        
        //output SET block
        sql.append(blockOfSet.get(0)).append(RTN_LINE);
        for(int i = 1; i < blockOfSet.size(); i++) {
            if(i > 1) {
                sql.append(", ");
            }
            sql.append(blockOfSet.get(i)).append(RTN_LINE);
        }
        
        sql.append(blockOfWhere.get(0)).append(RTN_LINE);
        for(int i = 1; i < blockOfWhere.size(); i++) {
            if (i > 1) {
                sql.append("AND "); 
            }
            sql.append(blockOfSet.get(i)).append(RTN_LINE);
        }

        sql.append(";").append(RTN_LINE);

        return sql.toString();
    }
    
    private final static String substituteColName(
            TableDesc tblDesc, String line) {
        int index0 = line.indexOf('@');
        if(index0 < 0) {
            return line;
        }
        
        int index1 = line.indexOf('=', index0);
        if(index1 < 0) {
            return line;
        }
        
        int colNum = Integer.parseInt(line.substring(index0 + 1, index1));
        return line.substring(0, index0)
                .concat("`").concat(tblDesc.getColNames()[colNum - 1]).concat("`")
                .concat(line.substring(index1))
                ;
    }
    
    private final static String findTableNameInLine(String line) {
        int index1 = line.lastIndexOf('`');
        int index0 = line.lastIndexOf('`', index1 - 1);
        
        return line.substring(index0 + 1, index1);
    }
    
    private final static TableDesc getTableDesc(TableDesc[] tableDescs, String tableName) {
        for(TableDesc desc : tableDescs) {
            if(desc.getTableName().equalsIgnoreCase(tableName)) {
                return desc;
            }
        }
        
        return null;
    }
    
    private final static boolean startsWithAndIgnoreSpace(String str, String prefix) {
        int begin;
        for(begin = 0; begin < str.length(); begin++) {
            char c = str.charAt(begin); 
            if(c != ' ') {
                break;
            }
        }
        
        return str.startsWith(prefix, begin);
    }

}

調用的例子：

    @Test
    public void test1() {
        try {
            File input = new File("test/restore001_#.sql");
            File output = new File("test/restore001_#formated.sql");
            
            BinLogSQLConverter.convertToSQL(
                    input, output, 
                    Charset.forName("utf-8"), 
                    new BinLogSQLConverter.TableDesc[] {
                        new BinLogSQLConverter.TableDesc(
                                "test_data1", 
                                new String[] {"code","name","col3","col4"}
                        ),
                        new BinLogSQLConverter.TableDesc(
                                "test_data2", 
                                new String[] {"code","name","col3","col4"}
                        ),
                    }
                    );
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }