首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >聊聊SpinalTap的Transaction

聊聊SpinalTap的Transaction

作者头像
code4it
发布于 2020-06-03 02:26:02
发布于 2020-06-03 02:26:02
37200
代码可运行
举报
文章被收录于专栏:码匠的流水账码匠的流水账
运行总次数:0
代码可运行

本文主要研究一下SpinalTap的Transaction

Transaction

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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
@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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
@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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
@Slf4j
@RequiredArgsConstructor
public abstract class MysqlMutationMapper<R extends BinlogEvent, T extends MysqlMutation>
    implements Mapper<R, List<T>> {
  @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 < row.length && columnIterator.hasNext(); position++) {
      final ColumnMetadata col = columnIterator.next();
      builder.put(col.getName(), new Column(col, row[position]));
    }

    return builder.build();
  }
}
  • 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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
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 < rows.size(); position++) {
      mutations.add(
          new MysqlInsertMutation(
              createMetadata(table, event, position),
              new Row(table, zip(rows.get(position), cols))));
    }

    return mutations;
  }
}
  • InsertMutationMapper继承了MysqlMutationMapper,其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

UpdateMutationMapper

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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
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[], Serializable[]>> rows = event.getRows();

    for (int position = 0; position < rows.size(); position++) {
      MysqlMutationMetadata metadata = createMetadata(table, event, position);

      final Row previousRow = new Row(table, zip(rows.get(position).getKey(), cols));
      final Row newRow = new Row(table, zip(rows.get(position).getValue(), cols));

      // If PK value has changed, then delete before image and insert new image
      // to retain invariant that a mutation captures changes to a single PK
      if (table.getPrimaryKey().isPresent()
          && !previousRow.getPrimaryKeyValue().equals(newRow.getPrimaryKeyValue())) {
        mutations.add(new MysqlDeleteMutation(metadata, previousRow));
        mutations.add(new MysqlInsertMutation(metadata, newRow));
      } else {
        mutations.add(new MysqlUpdateMutation(metadata, previousRow, newRow));
      }
    }

    return mutations;
  }
}
  • UpdateMutationMapper继承了MysqlMutationMapper,其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

DeleteMutationMapper

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

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
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 < rows.size(); position++) {
      mutations.add(
          new MysqlDeleteMutation(
              createMetadata(table, event, position),
              new Row(table, zip(rows.get(position), cols))));
    }

    return mutations;
  }
}
  • DeleteMutationMapper继承了MysqlMutationMapper,其构造器要求输入dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch参数

小结

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

doc

  • Transaction
本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-06-01,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 码匠的流水账 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
聊聊SpinalTap的MysqlEventFilter
SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/filter/MysqlEventFilter.java
code4it
2020/05/31
3330
聊聊SpinalTap的MysqlEventFilter
聊聊SpinalTap的BinlogEventListener
SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/binlog_connector/BinaryLogConnectorSource.java
code4it
2020/05/29
3080
聊聊SpinalTap的BinlogEventListener
聊聊SpinalTap的BinlogEvent
SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/BinlogEvent.java
code4it
2020/05/30
2200
聊聊SpinalTap的BinlogEvent
golang源码分析:go-mysql(4)binlog增量同步
实现binlog增量同步(Incremental dumping)需要哪些步骤呢?获取配置,初始化同步器,找到上一次同步位置,开启同步,并处理解析到的事件,整体流程如下:
golangLeetcode
2023/09/06
1.1K0
golang源码分析:go-mysql(4)binlog增量同步
聊聊maxwell的BinlogConnectorEventListener
本文主要研究一下maxwell的BinlogConnectorEventListener
code4it
2020/04/30
4040
聊聊maxwell的BinlogConnectorEventListener
聊聊debezium的eventHandlers
debezium-v1.1.1.Final/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/BinlogReader.java
code4it
2020/05/19
5550
聊聊debezium的eventHandlers
golang源码分析:go-mysql(2)自己实现一个canel
如何用golang自己实现一个canel呢,github.com/go-mysql-org/go-mysql给我们提供了这样的能力,它已经完成mysql协议的解析,并将解析后同步从库的过程实现,加入了几个插件点,实现自己的canel只需要实现这几个插件点即可完成我们自定义的同步工具。下面我们结合源码分析一下如何实现。实现一个canel需要下面四步:初始化配置,创建canel实例,设置事件处理函数,开始处理。
golangLeetcode
2023/09/06
9860
golang源码分析:go-mysql(2)自己实现一个canel
基于python的mysql复制工具
python-mysql-replication 是基于python实现的 MySQL复制协议工具,我们可以用它来解析binlog 获取日志的insert,update,delete等事件 ,并基于此做其他业务需求。比如数据更改时失效缓存,监听dml事件通知下游业务方做对应处理。
用户1278550
2019/07/01
2.6K0
基于python的mysql复制工具
聊聊BinlogConnectorReplicator的getTransactionRows
本文主要研究一下BinlogConnectorReplicator的getTransactionRows
code4it
2020/05/01
4280
聊聊rocketmq-mysql的EventProcessor
rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/binlog/EventProcessor.java
code4it
2020/05/24
7410
聊聊rocketmq-mysql的EventProcessor
聊聊maxwell的FileProducer
maxwell-1.25.1/src/main/java/com/zendesk/maxwell/producer/FileProducer.java
code4it
2020/05/16
5210
聊聊puma的Parser
puma/puma/src/main/java/com/dianping/puma/parser/Parser.java
code4it
2020/06/02
5770
聊聊puma的Parser
聊聊canal的Position
canal-1.1.4/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/Position.java
code4it
2020/04/19
9390
聊聊canal的Position
聊聊puma的DefaultTaskExecutor
puma/puma/src/main/java/com/dianping/puma/taskexecutor/TaskExecutor.java
code4it
2020/06/03
1.1K0
聊聊puma的DefaultTaskExecutor
聊聊BinaryLogClient的connect
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
code4it
2020/05/05
1.5K0
聊聊debezium的BinlogReader
debezium-v1.1.1.Final/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/Reader.java
code4it
2020/05/18
8200
聊聊debezium的BinlogReader
聊聊puma的Parser
puma/puma/src/main/java/com/dianping/puma/parser/Parser.java
code4it
2020/06/04
3640
聊聊puma的DefaultDataHandler
puma/puma/src/main/java/com/dianping/puma/datahandler/DataHandler.java
code4it
2020/06/04
3880
聊聊puma的DefaultDataHandler
如何使用Python模拟MySQL Slave,可以看看这个开源项目
在MySQL中通过Master向Slave推送binlog数据变化,从而实现主从复制的过程,是一件看似再正常不过的事情了。整个过程可以使用如下的流程图来表示。
jeanron100
2019/10/28
1.4K0
聊聊maxwell的Recovery
maxwell-1.25.1/src/main/java/com/zendesk/maxwell/recovery/Recovery.java
code4it
2020/05/09
6780
聊聊maxwell的Recovery
相关推荐
聊聊SpinalTap的MysqlEventFilter
更多 >
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档