创建一个数据库 test
create database test;
使用 test 数据库
use test;
创建一个表 test1
create table test1 (field1 varchar(10),field2 int);
创建一个表 test2
create table test2(field1 varchar(2),field2 varchar(2),field3 varchar(3),field4 varchar (4));
从 information_schema.COLUMNS 表中筛选出 TABLE_SCHEMA 是库名 test 的数据，根据 表名 和 序号进行排序
SELECT TABLE_NAME , COLUMN_NAME, DATA_TYPE, CHARACTER_SET_NAME, ORDINAL_POSITION, COLUMN_TYPE, COLUMN_KEY FROM `information_schema`.`COLUMNS` WHERE TABLE_SCHEMA = 'test' ORDER BY TABLE_NAME, ORDINAL_POSITION;
得到的数据中，列的名称是顺序的
+------------+-------------+-----------+--------------------+------------------+-------------+------------+
| TABLE_NAME | COLUMN_NAME | DATA_TYPE | CHARACTER_SET_NAME | ORDINAL_POSITION | COLUMN_TYPE | COLUMN_KEY |
+------------+-------------+-----------+--------------------+------------------+-------------+------------+
| test1 | field1 | varchar | latin1 | 1 | varchar(10) | |
| test1 | field2 | int | NULL | 2 | int(11) | |
| test2 | field1 | varchar | latin1 | 1 | varchar(2) | |
| test2 | field2 | varchar | latin1 | 2 | varchar(2) | |
| test2 | field3 | varchar | latin1 | 3 | varchar(3) | |
| test2 | field4 | varchar | latin1 | 4 | varchar(4) | |
+------------+-------------+-----------+--------------------+------------------+-------------+------------+
test2 中插入一条数据
insert into test2 values('l','zw','dp','19');
监听到的 binlog 时间如下
Event{header=EventHeaderV4{timestamp=1586509755000, eventType=EXT_WRITE_ROWS, serverId=1, headerLength=19, dataLength=28, nextPosition=2745, flags=0}, data=WriteRowsEventData{tableId=223, includedColumns={0, 1, 2, 3}, rows=[ [[B@1e80bfe8, [B@66a29884, [B@4769b07b, [B@cc34f4d] ]}}
includedColumns={0, 1, 2, 3} 表示有四列
rows=[ [[B@1e80bfe8, [B@66a29884, [B@4769b07b, [B@cc34f4d] ] 表示一行数据中
为了能够将字段值对应的字段名称取出
必须按照 `information_schema`.`COLUMNS` 查询的结果，遍历指定表的所有字段，同时维护一个索引标记，
用这个索引标记，从 includedColumns 中获取对应的波尔值，以判断，这个列在数据中是否存在
如果存在，使用索引标记 从表示单元格数据的数组中取得对应的值作为 value，当前 列的名称作为 key
如果不存在的话，把 null 当作 value，当前列的名称作为 key
添加到一个 map 中，以此来达到每隔字段值都有对应的名字
java 代码
private void writeData(Table table, RowMap row, Serializable[] data, BitSet includedColumns) {
int dataIdx = 0, colIdx = 0;
for ( ColumnDef cd : table.getColumnList() ) {
if ( includedColumns.get(colIdx) ) {
Object json = null;
if ( data[dataIdx] != null ) {
json = cd.asJSON(data[dataIdx]);
}
row.putData(cd.getName(), json);
dataIdx++;
}
colIdx++;
}
}
