聊聊SpinalTap的BinlogEventListener

本文主要研究一下 SpinalTap 的 BinlogEventListener

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/binlog_connector/BinaryLogConnectorSource.java

@Slf4j
public final class BinaryLogConnectorSource extends MysqlSource {

  //......

  private final class BinlogEventListener implements BinaryLogClient.EventListener {public void onEvent(Event event) {Preconditions.checkState(isStarted(), "Source is not started and should not process events");

      final EventHeaderV4 header = event.getHeader();
      final BinlogFilePos filePos =
          new BinlogFilePos(binlogClient.getBinlogFilename(),
              header.getPosition(),
              header.getNextPosition(),
              binlogClient.getGtidSet(),
              serverUUID);

      BinaryLogConnectorEventMapper.INSTANCE
          .map(event, filePos)
          .ifPresent(BinaryLogConnectorSource.super::processEvent);
    }
  }

  //......

}

SpinalTap 的 BinlogEventListener 实现了 BinaryLogClient.EventListener 接口，其 onEvent 方法创建 BinlogFilePos，然后使用 BinaryLogConnectorEventMapper.INSTANCE 进行转换，最后执行 BinaryLogConnectorSource 的 processEvent 方法

SpinalTap/spinaltap-model/src/main/java/com/airbnb/spinaltap/mysql/BinlogFilePos.java

@Slf4j
@Getter
@EqualsAndHashCode
@NoArgsConstructor
@JsonDeserialize(builder = BinlogFilePos.Builder.class)
public class BinlogFilePos implements Comparable<BinlogFilePos>, Serializable {
  private static final long serialVersionUID = 1549638989059430876L;

  private static final Splitter SPLITTER = Splitter.on(':');
  private static final String NULL_VALUE = "null";
  public static final String DEFAULT_BINLOG_FILE_NAME = "mysql-bin-changelog";

  @JsonProperty private String fileName;
  @JsonProperty private long position;
  @JsonProperty private long nextPosition;
  @JsonProperty private GtidSet gtidSet;
  @JsonProperty private String serverUUID;

  public BinlogFilePos(long fileNumber) {this(fileNumber, 4L, 4L);
  }

  public BinlogFilePos(String fileName) {this(fileName, 4L, 4L);
  }

  public BinlogFilePos(long fileNumber, long position, long nextPosition) {this(String.format("%s.%06d", DEFAULT_BINLOG_FILE_NAME, fileNumber), position, nextPosition);
  }

  public BinlogFilePos(String fileName, long position, long nextPosition, String gtidSet, String serverUUID) {
    this.fileName = fileName;
    this.position = position;
    this.nextPosition = nextPosition;
    this.serverUUID = serverUUID;
    if (gtidSet != null) {this.gtidSet = new GtidSet(gtidSet);
    }
  }

  public BinlogFilePos(String fileName, long position, long nextPosition) {this(fileName, position, nextPosition, null, null);
  }

  public static BinlogFilePos fromString(@NonNull final String position) {Iterator<String> parts = SPLITTER.split(position).iterator();
    String fileName = parts.next();
    String pos = parts.next();
    String nextPos = parts.next();

    if (NULL_VALUE.equals(fileName)) {fileName = null;}

    return new BinlogFilePos(fileName, Long.parseLong(pos), Long.parseLong(nextPos));
  }

  @JsonIgnore
  public long getFileNumber() {if (fileName == null) {return Long.MAX_VALUE;}
    if (fileName.equals("")) {return Long.MIN_VALUE;}
    String num = fileName.substring(fileName.lastIndexOf('.') + 1);
    return Long.parseLong(num);
  }

  @Override
  public String toString() {return String.format("%s:%d:%d", fileName, position, nextPosition);
  }

  @Override
  public int compareTo(@NonNull final BinlogFilePos other) {if (shouldCompareUsingFilePosition(this, other)) {return getFileNumber() != other.getFileNumber()
          ? Long.compare(getFileNumber(), other.getFileNumber())
          : Long.compare(getPosition(), other.getPosition());
    }

    if (this.gtidSet.equals(other.gtidSet)) {return 0;}
    if (this.gtidSet.isContainedWithin(other.gtidSet)) {return -1;}
    return 1;
  }

  /** Check if two BinlogFilePos are from the same source MySQL server */
  private static boolean isFromSameSource(BinlogFilePos pos1, BinlogFilePos pos2) {return pos1.getServerUUID() != null
        && pos1.getServerUUID().equalsIgnoreCase(pos2.getServerUUID());
  }

  /** Whether we can compare two BinlogFilePos using Binlog file position (without GTIDSet) */
  public static boolean shouldCompareUsingFilePosition(BinlogFilePos pos1, BinlogFilePos pos2) {return isFromSameSource(pos1, pos2) || pos1.getGtidSet() == null || pos2.getGtidSet() == null;
  }

  //......

}

BinlogFilePos 定义了 fileName、position、nextPosition、serverUUID 属性

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/binlog_connector/BinaryLogConnectorEventMapper.java

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class BinaryLogConnectorEventMapper {public static final BinaryLogConnectorEventMapper INSTANCE = new BinaryLogConnectorEventMapper();

  public Optional<BinlogEvent> map(@NonNull final Event event, @NonNull final BinlogFilePos position) {final EventHeaderV4 header = event.getHeader();
    final EventType eventType = header.getEventType();
    final long serverId = header.getServerId();
    final long timestamp = header.getTimestamp();

    if (EventType.isWrite(eventType)) {final WriteRowsEventData data = event.getData();
      return Optional.of(new WriteEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else if (EventType.isUpdate(eventType)) {final UpdateRowsEventData data = event.getData();
      return Optional.of(new UpdateEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else if (EventType.isDelete(eventType)) {final DeleteRowsEventData data = event.getData();
      return Optional.of(new DeleteEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else {switch (eventType) {
        case TABLE_MAP:
          TableMapEventData tableMapData = event.getData();
          return Optional.of(
              new TableMapEvent(tableMapData.getTableId(),
                  serverId,
                  timestamp,
                  position,
                  tableMapData.getDatabase(),
                  tableMapData.getTable(),
                  tableMapData.getColumnTypes()));
        case XID:
          final XidEventData xidData = event.getData();
          return Optional.of(new XidEvent(serverId, timestamp, position, xidData.getXid()));
        case GTID:
          final GtidEventData gtidEventData = event.getData();
          return Optional.of(new GTIDEvent(serverId, timestamp, position, gtidEventData.getGtid()));
        case QUERY:
          final QueryEventData queryData = event.getData();
          return Optional.of(
              new QueryEvent(
                  serverId,
                  timestamp,
                  position,
                  queryData.getDatabase(),
                  queryData
                      .getSql()
                      // https://dev.mysql.com/doc/refman/5.7/en/comments.html
                      // Replace MySQL-specific comments (/*! ... */ and /*!50110 ... */) which
                      // are actually executed
                      .replaceAll("/\\*!(?:\\d{5})?(.*?)\\*/", "$1")
                      // Remove block comments
                      // https://stackoverflow.com/questions/13014947/regex-to-match-a-c-style-multiline-comment
                      // line comments and newlines are kept
                      // Note: This does not handle comments in quotes
                      .replaceAll("/\\*[^*]*\\*+(?:[^/*][^*]*\\*+)*/", " ")
                      // Remove extra spaces
                      .replaceAll("\\h+", " ")
                      .replaceAll("^\\s+", "")));
        case FORMAT_DESCRIPTION:
          return Optional.of(new StartEvent(serverId, timestamp, position));
        default:
          return Optional.empty();}
    }
  }
}

BinaryLogConnectorEventMapper 提供了 map 方法，它接收 Event、BinlogFilePos 参数，然后根据 eventType 来返回不同的 BinlogEvent

SpinalTap/spinaltap-common/src/main/java/com/airbnb/spinaltap/common/source/AbstractSource.java

@Slf4j
@RequiredArgsConstructor
public abstract class AbstractSource<E extends SourceEvent> extends ListenableSource<E> {

  //......

  public final void processEvent(final E event) {
    try {if (!eventFilter.apply(event)) {log.debug("Event filtered from source {}. Skipping. event={}", name, event);
        return;
      }

      notifyEvent(event);

      final Stopwatch stopwatch = Stopwatch.createStarted();

      metrics.eventReceived(event);
      log.debug("Received event from source {}. event={}", name, event);

      notifyMutations(mutationMapper.map(event));

      stopwatch.stop();
      final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS);

      metrics.processEventTime(event, time);

    } catch (Exception ex) {if (!isStarted()) {
        // Do not process the exception if streaming has stopped.
        return;
      }

      final String errorMessage = String.format("Failed to process event from source %s", name);

      log.error(errorMessage, ex);
      metrics.eventFailure(ex);

      notifyError(ex);

      throw new SourceException(errorMessage, ex);
    }
  }

  //......

}

AbstractSource 的 processEvent 方法首先执行 eventFilter.apply(event)，没有被过滤掉的话则执行 notifyEvent 及 notifyMutations

SpinalTap/spinaltap-common/src/main/java/com/airbnb/spinaltap/common/source/ListenableSource.java

abstract class ListenableSource<E extends SourceEvent> implements Source {private final List<Listener> listeners = new ArrayList<>();

  @Override
  public void addListener(@NonNull final Listener listener) {listeners.add(listener);
  }

  @Override
  public void removeListener(@NonNull final Listener listener) {listeners.remove(listener);
  }

  protected void notifyMutations(final List<? extends Mutation<?>> mutations) {if (!mutations.isEmpty()) {listeners.forEach(listener -> listener.onMutation(mutations));
    }
  }

  protected void notifyEvent(E event) {listeners.forEach(listener -> listener.onEvent(event));
  }

  protected void notifyError(Throwable error) {listeners.forEach(listener -> listener.onError(error));
  }

  protected void notifyStart() {listeners.forEach(Source.Listener::onStart);
  }
}

ListenableSource 的 notifyEvent 及 notifyMutations 都是遍历 listeners 执行 listener.onEvent 或者 listener.onMutation 方法

SpinalTap/spinaltap-common/src/main/java/com/airbnb/spinaltap/common/pipe/Pipe.java

  final class SourceListener extends Source.Listener {public void onMutation(List<? extends Mutation<?>> mutations) {destination.send(mutations);
    }

    public void onError(Throwable error) {errorHandlingExecutor.execute(Pipe.this::close);
    }
  }

SourceListener 继承了 Source.Listener，其 onMutation 方法执行 destination.send(mutations)

SpinalTap 的 BinlogEventListener 实现了 BinaryLogClient.EventListener 接口，其 onEvent 方法创建 BinlogFilePos，然后使用 BinaryLogConnectorEventMapper.INSTANCE 进行转换，最后执行 BinaryLogConnectorSource 的 processEvent 方法

BinaryLogConnectorSource

聊聊SpinalTap的BinlogEventListener

序

BinlogEventListener

BinlogFilePos

BinaryLogConnectorEventMapper

AbstractSource

ListenableSource

SourceListener

小结

doc