MySQL 二進制日志 binlog 核心知識點小結
binlog的概念和基本作用
bin log實際上是一個物理日志，當我們對某個數據頁進行修改操作時我們就會將這個操作寫到bin log中，當我們數據庫需要進行主備、主從復制等操作時，都可以基于bin log保證數據一致性。
binlog緩沖區
bin log緩沖區和我們的redo log和undo log緩沖區有所不同，redo log和undo log緩存都在存儲引擎的共享緩沖區緩沖區buffer pool中，而bin log則是為每個工作線程獨立分配一個內存作為bin log緩沖區：
bin log之所以是在每個線程中，原因有二：
- 考慮到mysql-server要求所有存儲引擎對于bin.log都是兼容。
- 將bin_log_buffer設置在每個線程/事務中還能保證并發操作的性能，避免為各個線程爭搶臨界緩沖區的沖突導致并發性能下降。
binlog對應的三種記錄格式
row：這種格式主要用于保證數據實時性的，例如我們執行下面這段SQL
update table set time=now() where id=1;如果我們將其存到bin log之后很長一段時間才提交事務，那么時間就會有所延遲，所以MySQL為了保證數據實時性，就會將寫入bin log中的SQL用row格式，如下圖所示，可以看到row格式的SQL語句時間是當前時間的具體值，并且where條件寫死了當前條件列，確保數據實時一致性：
當然這樣做的缺點也很明顯，如果涉及大批量操作，那么針對每條數據對應的都會生成對應的row語句，那么對于內存的占用就很高，進行恢復和同步時的IO和SQL執行時間也是非常不友好的。
stament：這種同步策略即執行的SQL是什么，對應傳輸過去的時對應的語句就是什么樣的，這就會導致我們上文所說的一致性問題：
mixed：這種格式就是為了上述兩種方案的混合體，如果操作可能出現數據不一致問題則用row格式，反之使用stament格式。
binlog文件日志格式
我們可以通過下面這條SQL語句看到我們本地的bin log文件:
show binary logs;輸出結果如下所示，可以看到bin log的格式基本都是mysql-bin.0000xxx:
mysql-bin.001606 440052 No
mysql-bin.001607 111520 Nobinlog是如何完成寫入
當我們開始事務時，將修改寫入bin log cache中，一旦事務提交，就會將bin log通過write寫入到文件系統緩存的page cache中，然后根據我們配置的刷盤參數將cache內容調用操作系統內核方法fsync將結果寫入到bin log 物理文件中:
而調用系統函數fsync的實際是根據MySQL系統參數決定的，這個系統變量查詢SQL如下：
SHOW VARIABLES LIKE 'sync_binlog';而sync_binlog值分別三種:
- 0.當配置為了0時，每次事務提交都只會write，fsync調用時機是由系統決定的。
- 1.當配置設置為1時，每次事務提交都會調用fsync。
- N. 當配置為N，代表提交了N個事務之后就會將page cache中的數據通過fsync進行刷盤。
binlog和redolog的區別
這個問題我們可以從以下幾個場景來表述一下：
從使用場景來說：
- bin log常用于數據災備或數據同步到其他異構程序中的場景。
- redo log常用于故障恢復保證數據持久性。
從數據內容來說：
- redo log存儲的物理日志，即修改的數據內容，對應的redo block結構體針對各種偏移量和修改涉及的頁都有及其復雜的涉及，這里就不多做贅述。
- bin log則是記錄可以是statment語句也可以是原生修改的row，具體可以通過查看binlog_format知曉。
生成范圍:
- bin log是MySQL server生成的事務日志，任何存儲引擎都可以使用
- redo log只有innodb這個存儲引擎支持。
(實踐)基于flink cdc訂閱binlog同步數據
接下來我們就基于spring boot演示一下如何基于flink cdc訂閱bin.log完成db庫中的tb_1和tb_2的數據訂閱和同步：
之所以筆者使用flink cdc而不是canel大體有以下幾個原因：
- flink cdc支持全量和增量同步以及斷點續傳等功能，尤其是斷點續傳這一點對于需要保證異構數據庫的數據一致性是非常好的。
- 性能表現更出色，按照阿里云的說法：
我們將全增量一體化框架與 Debezium 1.6 版本做 簡單的 TPC-DS 讀取測試對比，customer 單表數據量 6500 萬，在 Flink CDC 用 8 個并發的情況下，吞吐提升了 6.8 倍，耗時僅 13 分鐘，得益于并發讀取的支持，如果用戶需要更快的讀取速度，用戶可以增加并發實現。
話不說我們給出基礎的集成步驟，首先是引入flink cdc和MySQL的依賴，這里筆者為了文章的簡練只給出的flink cdc相關的pom依賴：
<properties>
<flink.version>1.13.6</flink.version>
</properties>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<!--mysql -cdc-->
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>2.0.0</version>
</dependency>然后我們在yml或者properties文件中給出MySQL配置即可，然后我們聲明一個CdcInfo記錄從bin.log中同步的數據：
@Data
publicclass CdcInfo {
/**
* 變更前數據
*/
private JSONObject beforeData;
/**
* 變更后數據
*/
private JSONObject afterData;
private String operation;
/**
* binlog 文件名
*/
private String binLogName;
/**
* binlog當前讀取點位
*/
private Integer filePos;
/**
* 數據庫名
*/
private String dbName;
/**
* 表名
*/
private String tbName;
/**
* 變更時間
*/
private Long changeTime;
}然后我們編寫一個關于bin.log通知事件的監聽，針對flink cdc配置筆者都基于CommandLineRunner 這個拓展點完成配置，這里面涉及眾多的flink cdc配置參數，可以看到筆者的程序同步模式配置的是initial即啟動后會進行全量同步再進行增量同步，同時通過表達式db.tb_[1-2]+指明僅僅處理tb_1和tb_2表的數據更新變化。
@Component
publicclass MysqlCdcEventListener implements CommandLineRunner {
//數據接收器用于應用架構更改和將更改數據寫入外部系統
privatefinal CdcSink cdcSink;
public MysqlCdcEventListener(CdcSink cdcSink) {
this.cdcSink = cdcSink;
}
@Override
public void run(String... args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//設置并行度
env.setParallelism(Runtime.getRuntime().availableProcessors());
DebeziumSourceFunction<CdcInfo> debeziumSource = buildDebeziumSource();
DataStream<CdcInfo> streamSource = env
.addSource(debeziumSource, "mysql-source")
.setParallelism(1);
//將流數據交給
streamSource.addSink(cdcSink);
env.execute("mysql-stream-cdc");
}
/**
* 構造變更數據源
*/
private DebeziumSourceFunction<CdcInfo> buildDebeziumSource() {
Properties debeziumProperties = new Properties();
//設置快照為無鎖
debeziumProperties.put("snapshot.locking.mode", "none");
return MySqlSource.<CdcInfo>builder()
.hostname("xxxx")
.port(3306)
.databaseList("db")
//監聽db庫中的[1-2]表
.tableList("db.tb_[1-2]+")
.username("xxxx")
.password("xxxx")
//設置為 initial:在第一次啟動時對受監視的數據庫表執行初始快照，并繼續讀取最新的 binlog
.startupOptions(StartupOptions.initial())
//設置序列化配置
.deserializer(new MysqlDeserialization())
.serverTimeZone("GMT+8")
.debeziumProperties(debeziumProperties)
.build();
}
}- 關于這些配置的信息建議讀者移步官方文檔的說明:https://nightlies.apache.org/flink/flink-cdc-docs-release-3.0/zh/
- 相應的使用配置示例讀者也可以參考flink cdc對應的GitHub上的說明:https://github.com/gunnarmorling/flink-cdc-connectors
上文代碼示例中給出一個涉及反序列化生產CdcInfo的操作，筆者指明了MysqlDeserialization 這里也給出對應的源碼示例：
public class MysqlDeserialization implements DebeziumDeserializationSchema<CdcInfo> {
publicstaticfinal String TS_MS = "ts_ms";
publicstaticfinal String BIN_FILE = "file";
publicstaticfinal String POS = "pos";
publicstaticfinal String CREATE = "CREATE";
publicstaticfinal String BEFORE = "before";
publicstaticfinal String AFTER = "after";
publicstaticfinal String SOURCE = "source";
publicstaticfinal String UPDATE = "UPDATE";
@Override
public void deserialize(SourceRecord sourceRecord, Collector<CdcInfo> collector) {
//獲取bin.log訂閱到的信息
String topic = sourceRecord.topic();
String[] fields = topic.split("\\.");
String database = fields[1];
String tableName = fields[2];
Struct struct = (Struct) sourceRecord.value();
final Struct source = struct.getStruct(SOURCE);
CdcInfo tbCdcInfo = new CdcInfo();
//獲取前后變化數據
tbCdcInfo.setBeforeData(convert2JsonObj(struct, BEFORE));
tbCdcInfo.setAfterData(convert2JsonObj(struct, AFTER));
//5.獲取操作類型 CREATE UPDATE DELETE
Envelope.Operation operation = Envelope.operationFor(sourceRecord);
String type = operation.toString().toUpperCase();
tbCdcInfo.setOperation(type);
tbCdcInfo.setBinLogName(Optional.ofNullable(source.get(BIN_FILE)).map(Object::toString).orElse(""));
tbCdcInfo.setFilePos(Optional.ofNullable(source.get(POS)).map(x -> Integer.parseInt(x.toString())).orElse(0));
tbCdcInfo.setDbName(database);
tbCdcInfo.setTbName(tableName);
tbCdcInfo.setChangeTime(Optional.ofNullable(struct.get(TS_MS)).map(x -> Long.parseLong(x.toString())).orElseGet(System::currentTimeMillis));
//7.輸出數據
collector.collect(tbCdcInfo);
}
/**
* 從原始數據獲取出變更之前或之后的數據
*/
private JSONObject convert2JsonObj(Struct value, String fieldElement) {
Struct element = value.getStruct(fieldElement);
JSONObject jsonObject = new JSONObject();
if (element != null) {
Schema afterSchema = element.schema();
List<Field> fieldList = afterSchema.fields();
for (Field field : fieldList) {
Object afterValue = element.get(field);
jsonObject.put(field.name(), afterValue);
}
}
return jsonObject;
}
@Override
public TypeInformation<CdcInfo> getProducedType() {
return TypeInformation.of(CdcInfo.class);
}
}此時我們啟動程序后針對數據表進行修改操作就會收到數據消息的訂閱了：
訂閱到的數據:CdcInfo(beforeData={"id":1,"name":"xiaoming"}, afterData={"id":1,"name":"xiaoming1"}, operatinotallow=UPDATE, binLogName=binlog.000156, filePos=1256, dbName=db, tbName=tb_2, changeTime=1734622269654)
