当前位置：首页 > 文章列表 > 数据库 > MySQL > 聊聊SpinalTap的Transaction

聊聊SpinalTap的Transaction

来源：SegmentFault 2023-02-19 07:41:13 0浏览收藏

大家好，今天本人给大家带来文章《聊聊SpinalTap的Transaction》，文中内容主要涉及到MySQL，如果你对数据库方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

序

本文主要研究一下SpinalTap的Transaction

Transaction

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

@Value
@RequiredArgsConstructor
public class Transaction {
  private final long timestamp;
  private final long offset;
  private final BinlogFilePos position;
  private final String gtid;

  public Transaction(long timestamp, long offset, BinlogFilePos position) {
    this.timestamp = timestamp;
    this.offset = offset;
    this.position = position;
    this.gtid = null;
  }
}

Transaction定义了timestamp、offset、position、gtid属性

MysqlMutationMetadata

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

@Value
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
public class MysqlMutationMetadata extends Mutation.Metadata {
  private final DataSource dataSource;
  private final BinlogFilePos filePos;
  private final Table table;
  private final long serverId;
  private final Transaction beginTransaction;
  private final Transaction lastTransaction;

  /** The leader epoch of the node resource processing the event. */
  private final long leaderEpoch;

  /** The mutation row position in the given binlog event. */
  private final int eventRowPosition;

  public MysqlMutationMetadata(
      DataSource dataSource,
      BinlogFilePos filePos,
      Table table,
      long serverId,
      long id,
      long timestamp,
      Transaction beginTransaction,
      Transaction lastTransaction,
      long leaderEpoch,
      int eventRowPosition) {
    super(id, timestamp);

    this.dataSource = dataSource;
    this.filePos = filePos;
    this.table = table;
    this.serverId = serverId;
    this.beginTransaction = beginTransaction;
    this.lastTransaction = lastTransaction;
    this.leaderEpoch = leaderEpoch;
    this.eventRowPosition = eventRowPosition;
  }
}

MysqlMutationMetadata定义了dataSource、filePos、table、serverId、beginTransaction、lastTransaction、leaderEpoch、eventRowPosition属性

MysqlMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/MysqlMutationMapper.java

@Slf4j
@RequiredArgsConstructor
public abstract class MysqlMutationMapper<r extends binlogevent t mysqlmutation>
    implements Mapper<r list>> {
  @NonNull private final DataSource dataSource;
  @NonNull private final TableCache tableCache;
  @NonNull private final AtomicReference<transaction> beginTransaction;
  @NonNull private final AtomicReference<transaction> lastTransaction;
  @NonNull private final AtomicLong leaderEpoch;

  public static Mapper<binlogevent list extends mutation>>> create(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final SchemaTracker schemaTracker,
      @NonNull final AtomicLong leaderEpoch,
      @NonNull final AtomicReference<transaction> beginTransaction,
      @NonNull final AtomicReference<transaction> lastTransaction,
      @NonNull final MysqlSourceMetrics metrics) {
    final AtomicReference<string> gtid = new AtomicReference();
    return ClassBasedMapper.<binlogevent list extends mutation>>>builder()
        .addMapper(TableMapEvent.class, new TableMapMapper(tableCache))
        .addMapper(GTIDEvent.class, new GTIDMapper(gtid))
        .addMapper(QueryEvent.class, new QueryMapper(beginTransaction, gtid, schemaTracker))
        .addMapper(XidEvent.class, new XidMapper(lastTransaction, gtid, metrics))
        .addMapper(StartEvent.class, new StartMapper(dataSource, tableCache, metrics))
        .addMapper(
            UpdateEvent.class,
            new UpdateMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .addMapper(
            WriteEvent.class,
            new InsertMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .addMapper(
            DeleteEvent.class,
            new DeleteMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .build();
  }

  protected abstract List<t> mapEvent(@NonNull final Table table, @NonNull final R event);

  public List<t> map(@NonNull final R event) {
    Table table = tableCache.get(event.getTableId());

    return mapEvent(table, event);
  }

  MysqlMutationMetadata createMetadata(
      @NonNull final Table table, @NonNull final BinlogEvent event, final int eventPosition) {
    return new MysqlMutationMetadata(
        dataSource,
        event.getBinlogFilePos(),
        table,
        event.getServerId(),
        event.getOffset(),
        event.getTimestamp(),
        beginTransaction.get(),
        lastTransaction.get(),
        leaderEpoch.get(),
        eventPosition);
  }

  static ImmutableMap<string column> zip(
      @NonNull final Serializable[] row, @NonNull final Collection<columnmetadata> columns) {
    if (row.length != columns.size()) {
      log.error("Row length {} and column length {} don't match", row.length, columns.size());
    }

    final ImmutableMap.Builder<string column> builder = ImmutableMap.builder();
    final Iterator<columnmetadata> columnIterator = columns.iterator();

    for (int position = 0; position </columnmetadata></string></columnmetadata></string></t></t></binlogevent></string></transaction></transaction></binlogevent></transaction></transaction></r></r>

MysqlMutationMapper定义了dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch属性；它提供了createMetadata方法，它接收table、event、eventPosition参数返回新建的MysqlMutationMetadata

InsertMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/InsertMutationMapper.java

class InsertMutationMapper extends MysqlMutationMapper<writeevent mysqlinsertmutation> {
  InsertMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<transaction> beginTransaction,
      @NonNull final AtomicReference<transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<mysqlinsertmutation> mapEvent(
      @NonNull final Table table, @NonNull final WriteEvent event) {
    final List<serializable> rows = event.getRows();
    final List<mysqlinsertmutation> mutations = new ArrayList();
    final Collection<columnmetadata> cols = table.getColumns().values();

    for (int position = 0; position </columnmetadata></mysqlinsertmutation></serializable></mysqlinsertmutation></transaction></transaction></writeevent>

InsertMutationMapper继承了MysqlMutationMapper，其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

UpdateMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/UpdateMutationMapper.java

final class UpdateMutationMapper extends MysqlMutationMapper<updateevent mysqlmutation> {
  UpdateMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<transaction> beginTransaction,
      @NonNull final AtomicReference<transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<mysqlmutation> mapEvent(
      @NonNull final Table table, @NonNull final UpdateEvent event) {
    final List<mysqlmutation> mutations = Lists.newArrayList();
    final Collection<columnmetadata> cols = table.getColumns().values();
    final List<map.entry serializable>> rows = event.getRows();

    for (int position = 0; position </map.entry></columnmetadata></mysqlmutation></mysqlmutation></transaction></transaction></updateevent>

UpdateMutationMapper继承了MysqlMutationMapper，其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

DeleteMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/DeleteMutationMapper.java

final class DeleteMutationMapper extends MysqlMutationMapper<deleteevent mysqldeletemutation> {
  DeleteMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<transaction> beginTransaction,
      @NonNull final AtomicReference<transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<mysqldeletemutation> mapEvent(
      @NonNull final Table table, @NonNull final DeleteEvent event) {
    final Collection<columnmetadata> cols = table.getColumns().values();
    final List<mysqldeletemutation> mutations = new ArrayList();
    final List<serializable> rows = event.getRows();

    for (int position = 0; position </serializable></mysqldeletemutation></columnmetadata></mysqldeletemutation></transaction></transaction></deleteevent>

DeleteMutationMapper继承了MysqlMutationMapper，其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

小结

Transaction定义了timestamp、offset、position、gtid属性；MysqlMutationMetadata定义了dataSource、filePos、table、serverId、beginTransaction、lastTransaction、leaderEpoch、eventRowPosition属性