当前位置：首页 > 文章列表 > 数据库 > MySQL > Flink EOS如何防止外部系统乱入--两阶段提交源码

Flink EOS如何防止外部系统乱入--两阶段提交源码

来源：SegmentFault 2023-01-10 08:30:54 0浏览收藏

在数据库实战开发的过程中，我们经常会遇到一些这样那样的问题，然后要卡好半天，等问题解决了才发现原来一些细节知识点还是没有掌握好。今天golang学习网就整理分享《Flink EOS如何防止外部系统乱入--两阶段提交源码》，聊聊MySQL、大数据、flink，希望可以帮助到正在努力赚钱的你。

一、前言

根据维基百科的定义，两阶段提交(Two-phase Commit，简称2PC)是巨人们用来解决分布式系统架构下的所有节点在进行事务提交时保持一致性问题而设计的一种算法，也可称之为协议。
在Flink 1.4版本中，社区将两阶段提交协议中的公共逻辑进行提取和封装，发布了可供用户自定义实现特定方法来达到flink EOS特点的TwoPhaseCommitSinkFunction。本文基于Flink 1.12.4，和大家一起拜读Flink两阶段提交的源码。

二、2PC简介

1. 定义

根据维基百科的定义，两阶段提交可以归纳为一目的两角色三条件（即两个重要角色在三个可成立的条件下，实现最终的一个目的），具体如下：

一个最终目的

分布式系统架构下的所有节点在进行事务提交时要保持一致性（即要么全部成功，要么全部失败）

两个重要角色

        @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // this is like the pre-commit of a 2-phase-commit transaction
        // we are ready to commit and remember the transaction

        checkState(
                currentTransactionHolder != null,
                "bug: no transaction object when performing state snapshot");

        long checkpointId = context.getCheckpointId();
        LOG.debug(
                "{} - checkpoint {} triggered, flushing transaction '{}'",
                name(),
                context.getCheckpointId(),
                currentTransactionHolder);

        preCommit(currentTransactionHolder.handle);
        pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
        LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);

        currentTransactionHolder = beginTransactionInternal();
        LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);

        state.clear();
        state.add(
                new State(
                        this.currentTransactionHolder,
                        new ArrayList(pendingCommitTransactions.values()),
                        userContext));
    }




        /** 调用该方法可在事务内进行写值操作。 */
    protected abstract void invoke(TXN transaction, IN value, Context context) throws Exception;

    /** 调用该方法可以开启一个新事务。 */
    protected abstract TXN beginTransaction() throws Exception;

    /**
     * 调用该方法可预提交上一步创建的事务。
     * 注：预提交必须执行所有必要的步骤来为将来可能发生的提交准备事务。此后事务可能仍会中止，但底层实现必须确保对已预提交事务的提交调用始终成功
     * <p>Usually implementation involves flushing the data.
     */
    protected abstract void preCommit(TXN transaction) throws Exception;

    /**
     * 调用该方法正式提交预提交的事务。
     * 注：如果方法执行失败（即提交事务失败），Flink应用会重启，然后调用recoverAndCommit方法，重新提交该事务
     */
    protected abstract void commit(TXN transaction);

    /**
     * 执行失败后，调用该方法用来恢复事务提交操作。
     * 注：用户自定义实现必须确保该方法的调用最终会被执行成功。如果调用仍然失败，flink应用会被重启并且再次执行调用；反复执行，如果最终失败（用户重启策略配置的为一定的次数的重启），则会导致数据的丢失。
     * 另外，事务执行的顺序和他们创建时的顺序保持一致。
     */
    protected void recoverAndCommit(TXN transaction) {
        commit(transaction);
    }

    /** 调用该方法取消事务 */
    protected abstract void abort(TXN transaction);

    /** 执行失败后，取消被协调者拒绝的事务 */
    protected void recoverAndAbort(TXN transaction) {
        abort(transaction);
    }

    /**
     * 恢复用户上下文后的子类调用的回调函数，用来处理已经提交或者取消并且不会再处理的事务操作。
     */
    protected void finishRecoveringContext(Collection<txn> handledTransactions) {}</txn></p>

TwpPhaseCommintSinkFunction是一个抽象类实现了CheckpointedFunction接口和CheckpointListener接口。

实现CheckpointedFunction接口方法如下：

        @Override
        /** 
        * 父类中该方法的定义为当需要请求检查点的快照时可调用此方法。
        * 子类中的方法对检查点的操作进行了事务相关的耦合：
        * 1. 校验事务状态，调用该方法时事务不能为null。
        * 2. 预提交当前的事务并将该事务加入待提交事务列表，为后续正式提交做准备。
        * 3. 开启新的事务。
        * 4. 清空存储事务状态的列表，然后记录当前待提交的事务。
    */
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // this is like the pre-commit of a 2-phase-commit transaction
        // we are ready to commit and remember the transaction

          /** 1. 检查事务状态，如果事务对象为空，则抛异常 */
        checkState(
                currentTransactionHolder != null,
                "bug: no transaction object when performing state snapshot");

        long checkpointId = context.getCheckpointId();
        LOG.debug(
                "{} - checkpoint {} triggered, flushing transaction '{}'",
                name(),
                context.getCheckpointId(),
                currentTransactionHolder);
                /** 2.1 预提交当前的事务 --TODO 最终会调用实现类中preCommit的逻辑*/
        preCommit(currentTransactionHolder.handle);
          /** 2.2 记录当前预提交事务到待提交事务列表中 */
        pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
        LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);
                /** 3. 开启新的事务 */
        currentTransactionHolder = beginTransactionInternal();
        LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);
                /** 4. 清空记录事务状态的列表，并记录当前预提交的事务 */
        state.clear();
        state.add(
                new State(
                        this.currentTransactionHolder,
                        new ArrayList(pendingCommitTransactions.values()),
                        userContext));
    }


        /**
        * 调用该方法初始化检查点状态。
        * 子类中的初始化方法对事务进行了相关的耦合操作：
        * 1. 获取检查点的状态列表
        * 2. 循环状态列表，获取提交待提交的事务，并记录待提交的事务。
        * 3. 终止未提交的事务，并记录未提交的事务。
        * 4. 开启新的事务
        */
        @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
           /** 1. 获取检查点的状态列表 */
        state = context.getOperatorStateStore().getListState(stateDescriptor);

        boolean recoveredUserContext = false;
        if (context.isRestored()) {
            LOG.info("{} - restoring state", name());
              /** 2. 循环状态列表，获取提交待提交的事务，并记录待提交的事务 */
            for (State<txn context> operatorState : state.get()) {
                  // 获取用户上下文
                userContext = operatorState.getContext();
                  // 获取待提交的事务
                List<transactionholder>> recoveredTransactions =
                        operatorState.getPendingCommitTransactions();
                List<txn> handledTransactions = new ArrayList(recoveredTransactions.size() + 1);
                for (TransactionHolder<txn> recoveredTransaction : recoveredTransactions) {
                    // If this fails to succeed eventually, there is actually data loss
                    // 恢复并提交待提交的事务
                      recoverAndCommitInternal(recoveredTransaction);
                      // 记录待提交的事务
                    handledTransactions.add(recoveredTransaction.handle);
                    LOG.info("{} committed recovered transaction {}", name(), recoveredTransaction);
                }
                                /** 3. 终止未提交的事务，并记录未提交的事务 */
                {
                    TXN transaction = operatorState.getPendingTransaction().handle;
                    recoverAndAbort(transaction);
                    handledTransactions.add(transaction);
                    LOG.info(
                            "{} aborted recovered transaction {}",
                            name(),
                            operatorState.getPendingTransaction());
                }

                  /** 回收用户上下文配置 */
                if (userContext.isPresent()) {
                    finishRecoveringContext(handledTransactions);
                    recoveredUserContext = true;
                }
            }
        }

        // if in restore we didn't get any userContext or we are initializing from scratch
          // 如果在恢复中没有获取到用户上下文，则进行上下文初始化
        if (!recoveredUserContext) {
            LOG.info("{} - no state to restore", name());

            userContext = initializeUserContext();
        }
          // 情况待提交事务列表
        this.pendingCommitTransactions.clear();

          /** 4. 开启新事务 */
        currentTransactionHolder = beginTransactionInternal();
        LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);
    }
</txn></txn></transactionholder></txn>

实现CheckpointListener接口方法如下：

        /**
        * checkPoint完成之后会调用该方法，主要负责对预提交事务的正式提交。
        */
        @Override
    public final void notifyCheckpointComplete(long checkpointId) throws Exception {
       
          /** 1. 获取所有待提交的事务列表 */ 
        Iterator<map.entry transactionholder>>> pendingTransactionIterator =
                pendingCommitTransactions.entrySet().iterator();
        Throwable firstError = null;
                /** 2. 循环提交待提交事务列表中的事务 */
        while (pendingTransactionIterator.hasNext()) {
            Map.Entry<long transactionholder>> entry = pendingTransactionIterator.next();
            Long pendingTransactionCheckpointId = entry.getKey();
            TransactionHolder<txn> pendingTransaction = entry.getValue();
            // 只提交早于checkpointId的事务
              if (pendingTransactionCheckpointId > checkpointId) {
                continue;
            }

            LOG.info(
                    "{} - checkpoint {} complete, committing transaction {} from checkpoint {}",
                    name(),
                    checkpointId,
                    pendingTransaction,
                    pendingTransactionCheckpointId);
                        // 超时警告
            logWarningIfTimeoutAlmostReached(pendingTransaction);
            try {
                  // 第二阶段提交事务
                commit(pendingTransaction.handle);
            } catch (Throwable t) {
                if (firstError == null) {
                    firstError = t;
                }
            }

            LOG.debug("{} - committed checkpoint transaction {}", name(), pendingTransaction);
                        // 从待提交事务列表中移除已经提交过的事务
            pendingTransactionIterator.remove();
        }

        if (firstError != null) {
            throw new FlinkRuntimeException(
                    "Committing one of transactions failed, logging first encountered failure",
                    firstError);
        }
    }

    @Override
    public void notifyCheckpointAborted(long checkpointId) {}</txn></long></map.entry>