tencent cloud

message Entry { //Entry 是单个订阅事件结构，一个事件相当于 MySQL 的一个 binlog event
 Header header = 1;  //事件头
 Event event   = 2;  //事件体
}


message Header {
 int32       version        = 1;   //Entry 协议版本
 SourceType  sourceType     = 2;   //源库的类型信息，包括 MySQL，Oracle 等类型
 MessageType messageType    = 3;   //消息的类型，也就是 Event 的类型，包括 BEGIN、COMMIT、DML 等
 uint32 timestamp           = 4;   //Event 在原始 binlog 中的时间戳
 int64  serverId            = 5;   //源的 serverId
 string fileName            = 6;   //源 binlog 的文件名称
 uint64 position            = 7;   //事件在源 binlog 文件中的偏移量
 string gtid                = 8;   //当前事务的 gtid
 string schemaName          = 9;   //变更影响的 schema
 string tableName           = 10;  //变更影响的 table
 uint64 seqId               = 11;  //全局递增序列号
 uint64 eventIndex          = 12;  //如果大的 event 分片，每个分片从0开始编号，当前版本无意义，留待后续扩展用
 bool   isLast              = 13;  //当前 event 是否 event 分片的最后一块，是则为 true，当前版本无意义，留待后续扩展用
 repeated KVPair properties = 15;
}


message Event {
 BeginEvent      beginEvent      = 1;  //binlog 中的 begin 事件
 DMLEvent        dmlEvent        = 2;  //binlog 中的 dml 事件
 CommitEvent     commitEvent     = 3;  //binlog 中的 commit 事件
 DDLEvent        ddlEvent        = 4;  //binlog 中的 ddl 事件
 RollbackEvent   rollbackEvent   = 5;  //rollback 事件，当前版本无意义
 HeartbeatEvent  heartbeatEvent  = 6;  //源库定时发送的心跳事件
 CheckpointEvent checkpointEvent = 7;  //订阅后台添加的 checkpoint 事件，每10秒自动生成一个，用于 Kafka 生产和消费位点管理
 repeated KVPair properties      = 15;
}

message Entries {
     repeated Entry items = 1; //entry list
}

message Envelope {
     int32  version                  = 1; //protocol version, 决定了 data 内容如何解码
     uint32 total                    = 2;
     uint32 index                    = 3;
     bytes  data                     = 4; //当前 version 为1, 表示 data 中数据为 Entries 被 PB 序列化之后的结果
     repeated KVPair properties      = 15;
}

// 创建一个 Flink-kafka 消费者
FlinkKafkaConsumer<RecordMsgObject> consumer = 
                       new FlinkKafkaConsumer<>(topic, new DeserializeProtobufToRecordMsgObject(), props);


                       
// 自定义反序列化器，将原始消息反序列化为 RecordMsgObject 对象  
@Override
public RecordMsgObject deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    RecordMsgObject obj = new RecordMsgObject();
    obj.topic = record.topic();
    obj.partition = record.partition();
    obj.offset = record.offset();
    obj.partitionSeq = getPartitionSeq(record);
    obj.key = new String(record.key());
    obj.headers = record.headers();
    // 这里收到的就是 Envelope 的二进制的值
    obj.value = record.value();
    return obj;
}           

//将收到的消息按照 parttition 分区处理
stream.keyBy(RecordMsgObject::getPartition)
      .process(new SubscribeMsgProcess(trans2sql)).setParallelism(1);

// 反序列出 envlope，本 demo 默认一个 envlope 就可以存储完整个 bnlog-event 的数据
SubscribeProtobufData.Envelope envelope = SubscribeProtobufData.Envelope.parseFrom(record.value);
if (1 != envelope.getVersion()) {
    throw new IllegalStateException(String.format("unsupported version: %d", envelope.getVersion()));
}

// 反序列得到 Entries
ByteString envelopeData = envelope.getData();
SubscribeProtobufData.Entries entries;
if (1 == envelope.getTotal()) {
    entries = SubscribeProtobufData.Entries.parseFrom(envelopeData.toByteArray());
} else {
    entries = SubscribeProtobufData.Entries.parseFrom(shardMsgMap.get(shardId).toByteArray());
    shardMsgMap.remove(shardId);
}

// 遍历每个 Entry,根据 Entry 类型去打印 sql
for (SubscribeProtobufData.Entry entry : entries.getItemsList()) {
    onEntry(record.partition, record.offset, ps, entry, trans2sql);
}