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