Debezium-BinaryLogClient

文章目录

  

• 
  
  
  
  • 概要
    
  • 核心流程
    
  • 技能名词解释
    
  • 技能细节
    
  • 小结
    
  
  

  
概要

BinaryLogClient类，用于毗连和监听 MySQL 服务器的二进制日志（binlog）
核心流程

技能名词解释

   ### GTID (Global Transaction Identifier) 明白
  #### 定义
GTID（Global Transaction Identifier）是 MySQL 从 5.6 版本开始引入的一种全局事务标识符。每个 GTID 在整个 MySQL 集群中都是唯一的，用于唯一标识一个事务。
  #### 格式
GTID 的格式通常为 `source_id:transaction_id`，此中：
- **source_id**：表示生成事务的 MySQL 实例的唯一标识符，通常是实例的 `server_id`。
- **transaction_id**：表示在该实例上实行的事务的顺序号。
  例如，`3E11FA47-71CA-11E1-9E33-C80AA9429562:23` 表示在 `server_id` 为 `3E11FA47-71CA-11E1-9E33-C80AA9429562` 的 MySQL 实例上实行的第 23 个事务。
  #### 重要用途
1. **事务跟踪**：
   - GTID 可以资助跟踪事务在主从复制中的传播环境。每个事务在主库上生成后，会被分配一个唯一的 GTID，并在从库上应用时保存相同的 GTID。
  2. **简化复制管理**：
   - 利用 GTID 可以简化复制配置和管理。例如，可以通过指定 GTID 范围来同步特定的事务，而不需要手动管理二进制日志文件和位置。
  3. **故障恢复**：
   - 在主从切换或故障恢复时，GTID 可以确保从库不会重复应用同一个事务，从而避免数据不一致的标题。
  4. **并行复制**：
   - GTID 支持并行复制，即多个线程可以同时应用差别的事务，进步复制性能。
  #### 配置
启用 GTID 复制需要在 MySQL 配置文件中设置以下参数：
- `gtid_mode=ON`：启用 GTID 模式。
- `enforce_gtid_consistency=ON`：逼迫 GTID 一致性，确保所有事务都可以被正确地跟踪和应用。
  #### 总结
GTID 是 MySQL 中用于唯一标识事务的全局标识符，有助于简化复制管理和故障恢复。通过 GTID，可以更方便地跟踪和管理事务在主从复制中的传播，确保数据的一致性和可靠性。
  技能细节

1. /**
2. * 建立与MySQL服务器的连接并初始化二进制日志复制所需设置。
3. *
4. * 该方法首先检查当前客户端是否已连接，如果已连接，则抛出IllegalStateException。
5. * 如果未连接，它将尝试使用指定的主机名和端口连接到MySQL服务器。
6. * 连接成功后，接收服务器的欢迎包，并进行身份验证。
7. * 如果未指定二进制日志文件名，则获取二进制日志文件名和位置。
8. * 检查并确认支持的校验和类型，请求二进制日志流。
9. * 最后，通知生命周期监听器连接成功，并启动保持活动线程（如果启用）。
10. * 监听事件包。
11. */
12. public void connect() throws IOException {
13.     if (this.connected) {
14.         throw new IllegalStateException("BinaryLogClient 已经连接");
15.     } else {
16.         GreetingPacket greetingPacket;
17.         IOException e;
18.         try {
19.             try {
20.                 // 创建Socket连接
21.                 Socket socket = this.socketFactory != null ? this.socketFactory.createSocket() : new Socket();
22.                 socket.connect(new InetSocketAddress(this.hostname, this.port));
23.                 this.channel = new PacketChannel(socket);
24.                 if (this.channel.getInputStream().peek() == -1) {
25.                     throw new EOFException();
26.                 }
27.             } catch (IOException var7) {
28.                 e = var7;
29.                 throw new IOException("连接到 MySQL " + this.hostname + ":" + this.port + " 失败。请确保其正在运行。", e);
30.             }
32.             // 接收欢迎包并进行身份验证
33.             greetingPacket = this.receiveGreeting();
34.             this.authenticate(greetingPacket.getScramble(), greetingPacket.getServerCollation());
36.             // 获取二进制日志文件名和位置
37.             if (this.binlogFilename == null) {
38.                 this.fetchBinlogFilenameAndPosition();
39.             }
41.             // 调整二进制日志位置
42.             if (this.binlogPosition < 4L) {
43.                 if (this.logger.isLoggable(Level.WARNING)) {
44.                     this.logger.warning("二进制日志位置从 " + this.binlogPosition + " 调整为 " + 4);
45.                 }
46.                 this.binlogPosition = 4L;
47.             }
49.             // 获取并确认支持的校验和类型
50.             ChecksumType checksumType = this.fetchBinlogChecksum();
51.             if (checksumType != ChecksumType.NONE) {
52.                 this.confirmSupportOfChecksum(checksumType);
53.             }
55.             // 请求二进制日志流
56.             this.requestBinaryLogStream();
57.         } catch (IOException var10) {
58.             e = var10;
59.             if (this.channel != null && this.channel.isOpen()) {
60.                 this.channel.close();
61.             }
62.             throw e;
63.         }
65.         // 设置连接状态并记录日志
66.         this.connected = true;
67.         if (this.logger.isLoggable(Level.INFO)) {
68.             this.logger.info("连接到 " + this.hostname + ":" + this.port + " at " + this.binlogFilename + "/" + this.binlogPosition + " (sid:" + this.serverId + ", cid:" + greetingPacket.getThreadId() + ")");
69.         }
71.         // 通知生命周期监听器连接成功
72.         synchronized(this.lifecycleListeners) {
73.             Iterator i$ = this.lifecycleListeners.iterator();
74.             while(i$.hasNext()) {
75.                 LifecycleListener lifecycleListener = (LifecycleListener)i$.next();
76.                 lifecycleListener.onConnect(this);
77.             }
78.         }
80.         // 启动保持活动线程（如果启用）
81.         if (this.keepAlive && !this.isKeepAliveThreadRunning()) {
82.             this.spawnKeepAliveThread();
83.         }
85.         // 确保事件数据反序列化器
86.         this.ensureEventDataDeserializer(EventType.ROTATE, RotateEventDataDeserializer.class);
87.         synchronized(this.gtidSetAccessLock) {
88.             if (this.gtidSet != null) {
89.                 this.ensureEventDataDeserializer(EventType.GTID, GtidEventDataDeserializer.class);
90.             }
91.         }
93.         // 监听事件包
94.         this.listenForEventPackets();
95.     }
96. }

复制代码

1. /**
2. * 监听事件数据包的方法
3. * 该方法用于持续监听来自服务器的事件数据包，并进行相应的处理
4. * 当检测到数据包时，会根据数据包的类型进行处理，包括错误处理和事件处理
5. * 如果连接断开或者出现异常，将停止监听并进行相应的异常处理
6. *
7. * @throws IOException 如果在读取数据流时发生I/O错误
8. */
9. private void listenForEventPackets() throws IOException {
10.     // 获取输入流，用于读取服务器发送的数据
11.     ByteArrayInputStream inputStream = this.channel.getInputStream();
13.     // 无限循环，持续监听事件数据包
14.     label202:
15.     while(true) {
16.         try {
17.             // 检查输入流是否有数据可读
18.             if (inputStream.peek() != -1) {
19.                 // 读取数据包长度
20.                 int packetLength = inputStream.readInteger(3);
21.                 // 跳过1字节的填充
22.                 inputStream.skip(1L);
23.                 // 读取标记字节，用于判断数据包类型
24.                 int marker = inputStream.read();
25.                 // 如果标记为255，表示接收到的是错误数据包
26.                 if (marker == 255) {
27.                     // 解析错误数据包并抛出异常
28.                     ErrorPacket errorPacket = new ErrorPacket(inputStream.read(packetLength - 1));
29.                     throw new ServerException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(), errorPacket.getSqlState());
30.                 }
32.                 // 解析事件数据包
33.                 Event event;
34.                 try {
35.                     // 根据数据包长度决定是否需要分块读取
36.                     event = this.eventDeserializer.nextEvent(packetLength == 16777215 ? new ByteArrayInputStream(this.readPacketSplitInChunks(inputStream, packetLength - 1)) : inputStream);
37.                 } catch (Exception var20) {
38.                     // 处理解析异常
39.                     Exception e = var20;
40.                     Throwable cause = e instanceof EventDataDeserializationException ? e.getCause() : e;
41.                     // 根据异常类型进行不同处理
42.                     if (!(cause instanceof EOFException) && !(cause instanceof SocketException)) {
43.                         if (!this.isConnected()) {
44.                             continue;
45.                         }
47.                         // 通知生命周期监听器解包失败
48.                         synchronized(this.lifecycleListeners) {
49.                             Iterator i$ = this.lifecycleListeners.iterator();
51.                             while(true) {
52.                                 if (!i$.hasNext()) {
53.                                     continue label202;
54.                                 }
56.                                 LifecycleListener lifecycleListener = (LifecycleListener)i$.next();
57.                                 lifecycleListener.onEventDeserializationFailure(this, e);
58.                             }
59.                         }
60.                     }
62.                     throw e;
63.                 }
65.                 // 处理解析成功的事件
66.                 if (this.isConnected()) {
67.                     this.notifyEventListeners(event);
68.                     this.updateClientBinlogFilenameAndPosition(event);
69.                     this.updateGtidSet(event);
70.                 }
71.                 continue;
72.             }
73.         } catch (Exception var21) {
74.             // 处理通信异常
75.             Exception e = var21;
76.             if (this.isConnected()) {
77.                 synchronized(this.lifecycleListeners) {
78.                     Iterator i$ = this.lifecycleListeners.iterator();
80.                     while(i$.hasNext()) {
81.                         LifecycleListener lifecycleListener = (LifecycleListener)i$.next();
82.                         lifecycleListener.onCommunicationFailure(this, e);
83.                     }
84.                 }
85.             }
86.         } finally {
87.             // 确保在结束监听时断开连接
88.             if (this.isConnected()) {
89.                 this.disconnectChannel();
90.             }
92.         }
94.         // 结束方法
95.         return;
96.     }
97. }

复制代码

1. /**
2. * 通知事件监听器
3. * 当有事件发生时，此方法会被调用以通知所有注册的事件监听器
4. * 如果事件的数据是EventDataWrapper类型，则会用外部事件数据替换事件数据
5. *
6. * @param event 发生的事件，用于通知监听器
7. */
8. private void notifyEventListeners(Event event) {
9.     // 检查事件数据是否为EventDataWrapper类型，如果是，则用外部事件数据替换事件数据
10.     if (event.getData() instanceof EventDeserializer.EventDataWrapper) {
11.         event = new Event(event.getHeader(), ((EventDeserializer.EventDataWrapper)event.getData()).getExternal());
12.     }
14.     // 同步eventListeners以确保线程安全
15.     synchronized(this.eventListeners) {
16.         // 遍历所有事件监听器
17.         Iterator i$ = this.eventListeners.iterator();
19.         while(i$.hasNext()) {
20.             EventListener eventListener = (EventListener)i$.next();
22.             try {
23.                 // 通知事件监听器处理事件
24.                 eventListener.onEvent(event);
25.             } catch (Exception var7) {
26.                 Exception e = var7;
27.                 // 如果日志级别为WARNING，记录异常信息
28.                 if (this.logger.isLoggable(Level.WARNING)) {
29.                     this.logger.log(Level.WARNING, eventListener + " choked on " + event, e);
30.                 }
31.             }
32.         }
34.     }
35. }

复制代码

1. /**
2. * 将事件添加到队列中以进行后续批量处理。
3. *
4. * @param event 从二进制日志中读取的事件
5. */
6. protected void enqueue(Event event) {
7.     // 检查事件是否为空，避免空指针异常
8.     if (event != null) {
9.         try {
10.             // 将事件放入队列中
11.             events.put(event);
12.         } catch (InterruptedException e) {
13.             // 处理中断异常，恢复中断状态并抛出连接异常
14.             Thread.interrupted();
15.             throw new ConnectException("在等待将事件添加到队列时被中断", e);
16.         }
17.     }
18. }

复制代码

1. /**
2. * 覆盖 poll 方法以从 MySQL 服务器获取并处理事件。
3. * 该方法会持续轮询事件，处理这些事件，并返回处理后的记录列表。
4. *
5. * @return 处理后的 SourceRecord 列表
6. */
7. @Override
8. public List<SourceRecord> poll() throws InterruptedException {
9.     logger.trace("从 MySQL 服务器 '{}' 轮询事件", serverName);
10.     while (running.get() && (events.drainTo(batchEvents, maxBatchSize - batchEvents.size()) == 0 || batchEvents.isEmpty())) {
11.         // 没有事件需要处理，因此暂停一段时间 ...
12.         metronome.pause();
13.     }
14.     logger.trace("准备从 MySQL 服务器 '{}' 处理 {} 个事件", events.size(), serverName);
16.     // 至少有一些记录需要处理 ...
17.     List<SourceRecord> records = new ArrayList<>(batchEvents.size());
18.     while (!batchEvents.isEmpty()) {
19.         Event event = batchEvents.poll();
20.         if (event == null) continue;
22.         // 更新源偏移信息 ...
23.         EventHeader eventHeader = event.getHeader();
24.         EventType eventType = eventHeader.getEventType();
25.         if (eventType == EventType.ROTATE) {
26.             EventData eventData = event.getData();
27.             RotateEventData rotateEventData;
28.             if (eventData instanceof EventDeserializer.EventDataWrapper) {
29.                 rotateEventData = (RotateEventData) ((EventDeserializer.EventDataWrapper) eventData).getInternal();
30.             } else {
31.                 rotateEventData = (RotateEventData) eventData;
32.             }
33.             source.setBinlogFilename(rotateEventData.getBinlogFilename());
34.             source.setBinlogPosition(rotateEventData.getBinlogPosition());
35.             source.setRowInEvent(0);
36.         } else if (eventHeader instanceof EventHeaderV4) {
37.             EventHeaderV4 trackableEventHeader = (EventHeaderV4) eventHeader;
38.             long nextBinlogPosition = trackableEventHeader.getNextPosition();
39.             if (nextBinlogPosition > 0) {
40.                 source.setBinlogPosition(nextBinlogPosition);
41.                 source.setRowInEvent(0);
42.             }
43.         }
44.         
45.         if (!running.get()) break;
47.         // 如果有处理此事件的处理器，将事件转发给它 ...
48.         EventHandler handler = eventHandlers.get(eventType);
49.         if (handler != null) {
50.             handler.handle(event, source, records::add);
51.         }
52.     }
53.     logger.trace("完成从 MySQL 服务器 '{}' 处理 {} 个事件", serverName);
55.     if (!this.running.get()) {
56.         // 应该停止，因此返回已经处理的记录，以防止在 DB 历史已停止的情况下持久化记录 ...
57.         return null;
58.     }
60.     // 已经处理完所有事件，清空批处理队列并返回记录 ...
61.     assert batchEvents.isEmpty();
62.     return records;
63. }

复制代码

小结

   ### Debezium 监听和处理 Binlog 事故的简要总结
  1. **轮询事故**：
   - `poll` 方法通过 `logger.trace` 记载开始从 MySQL 服务器轮询事故。
   - 利用 `while` 循环不断查抄是否有新的事故需要处理。如果没有事故，调用 `metronome.pause()` 停息一段时间。
  2. **准备事故**：
   - 当有事故可用时，记载准备处理的事故数目。
   - 创建一个 `List<SourceRecord>` 来存储处理后的记载。
  3. **处理事故**：
   - 从 `batchEvents` 队列中取失事故并举行处理。
   - 根据事故范例更新源偏移信息：
     - 对于 `ROTATE` 事故，更新二进制日志文件名和位置。
     - 对于其他范例的事故，更新二进制日志位置。
   - 如果有相应的事故处理器，调用 `handler.handle` 方法处理事故并将结果添加到 `records` 列表中。
  4. **制止处理**：
   - 如果 `running` 标志为 `false`，表示应该制止处理，返回 `null` 以防止在 DB 汗青已制止的环境下长期化记载。
  5. **返回结果**：
   - 清空 `batchEvents` 队列，确保所有事故都已处理完毕。
   - 返回处理后的 `records` 列表。
  ### 关键步骤总结
- **轮询和等待**：通过循环和停息机制等待新事故。
- **事故处理**：根据事故范例更新偏移信息，并调用相应的处理器处理事故。
- **制止机制**：在需要制止时返回 `null`，避免不须要的记载长期化。
- **结果返回**：清空批处理队列并返回处理后的记载列表。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。