当前位置：首页 > 文章列表 > 数据库 > 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
    implements Mapper> {
  @NonNull private final DataSource dataSource;
  @NonNull private final TableCache tableCache;
  @NonNull private final AtomicReference beginTransaction;
  @NonNull private final AtomicReference lastTransaction;
  @NonNull private final AtomicLong leaderEpoch;

  public static Mapper>> create(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final SchemaTracker schemaTracker,
      @NonNull final AtomicLong leaderEpoch,
      @NonNull final AtomicReference beginTransaction,
      @NonNull final AtomicReference lastTransaction,
      @NonNull final MysqlSourceMetrics metrics) {
    final AtomicReference gtid = new AtomicReference();
    return ClassBasedMapper.>>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 mapEvent(@NonNull final Table table, @NonNull final R event);

  public List 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 zip(
      @NonNull final Serializable[] row, @NonNull final Collection columns) {
    if (row.length != columns.size()) {
      log.error("Row length {} and column length {} don't match", row.length, columns.size());
    }

    final ImmutableMap.Builder builder = ImmutableMap.builder();
    final Iterator columnIterator = columns.iterator();

    for (int position = 0; position

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 {
  InsertMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference beginTransaction,
      @NonNull final AtomicReference lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

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

    for (int position = 0; position

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 {
  UpdateMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference beginTransaction,
      @NonNull final AtomicReference lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

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

    for (int position = 0; position

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 {
  DeleteMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference beginTransaction,
      @NonNull final AtomicReference lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

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

    for (int position = 0; position

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

小结

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