概述

最近Flink母公司Data Artisans发布了一篇博客关于一个新的组件Streaming Ledger，给出了流数据的事务解决方案（就是常说的数据库的事务，满足ACID，隔离级别为Serializable）。

使用姿势

• 举例使用经典的转账和存款问题
  • 它是基于Flink的，关于Flink任务初始化的一些内容就不放在这里了
• 首先创建StreamingLedger。

        // start building the transactional streams
        StreamingLedger tradeLedger = StreamingLedger.create("simple trade example");

• 第二定义所需要用到的状态和相应的KV类型，这里分别是账户和账单明细。

        // define transactors on states
        tradeLedger.usingStream(deposits, "deposits")
                .apply(new DepositHandler())
                .on(accounts, DepositEvent::getAccountId, "account", READ_WRITE)
                .on(books, DepositEvent::getBookEntryId, "asset", READ_WRITE);

• 第三步是分别将输入流，具体的事务操作，操作的状态、从事件中获取key的方法、别名（会在事务操作，即此处的TxnHandler中体现）、和权限分别注入StreamLedger并输出为一个sideOutput。

        // produce transactions stream
        DataStream<TransactionEvent> transfers = env.addSource(new TransactionsGenerator(1));

        OutputTag<TransactionResult> transactionResults = tradeLedger.usingStream(transfers, "transactions")
                .apply(new TxnHandler())
                .on(accounts, TransactionEvent::getSourceAccountId, "source-account", READ_WRITE)
                .on(accounts, TransactionEvent::getTargetAccountId, "target-account", READ_WRITE)
                .on(books, TransactionEvent::getSourceBookEntryId, "source-asset", READ_WRITE)
                .on(books, TransactionEvent::getTargetBookEntryId, "target-asset", READ_WRITE)
                .output();

• 第四步是根据sideOuput的OutputTag输出结果，到这里，除了TxnHandler需要去实现以外，主干逻辑已经完成了。

        //  compute the resulting streams.
        ResultStreams resultsStreams = tradeLedger.resultStreams();

        // output to the console
        resultsStreams.getResultStream(transactionResults).print();

• 最后就是实现TxnHandler, 具体的转账和写入明细的逻辑都在这里。值得注意的是状态的获取依赖于上一步中在StreamLedger注入的别名，更新完状态之后再输出。

    private static final class TxnHandler extends TransactionProcessFunction<TransactionEvent, TransactionResult> {

        private static final long serialVersionUID = 1;

        @ProcessTransaction
        public void process(
                final TransactionEvent txn,
                final Context<TransactionResult> ctx,
                final @State("source-account") StateAccess<Long> sourceAccount,
                final @State("target-account") StateAccess<Long> targetAccount,
                final @State("source-asset") StateAccess<Long> sourceAsset,
                final @State("target-asset") StateAccess<Long> targetAsset) {

            final long sourceAccountBalance = sourceAccount.readOr(ZERO);
            final long sourceAssetValue = sourceAsset.readOr(ZERO);
            final long targetAccountBalance = targetAccount.readOr(ZERO);
            final long targetAssetValue = targetAsset.readOr(ZERO);

            // check the preconditions
            if (sourceAccountBalance > txn.getMinAccountBalance()
                    && sourceAccountBalance > txn.getAccountTransfer()
                    && sourceAssetValue > txn.getBookEntryTransfer()) {

                // compute the new balances
                final long newSourceBalance = sourceAccountBalance - txn.getAccountTransfer();
                final long newTargetBalance = targetAccountBalance + txn.getAccountTransfer();
                final long newSourceAssets = sourceAssetValue - txn.getBookEntryTransfer();
                final long newTargetAssets = targetAssetValue + txn.getBookEntryTransfer();

                // write back the updated values
                sourceAccount.write(newSourceBalance);
                targetAccount.write(newTargetBalance);
                sourceAsset.write(newSourceAssets);
                targetAsset.write(newTargetAssets);

                // emit result event with updated balances and flag to mark transaction as processed
                ctx.emit(new TransactionResult(txn, true, newSourceBalance, newTargetBalance));
            }
            else {
                // emit result with unchanged balances and a flag to mark transaction as rejected
                ctx.emit(new TransactionResult(txn, false, sourceAccountBalance, targetAccountBalance));
            }
        }
    }

原理

• 其实我的第一想法是，卧槽好牛逼，这得涉及到分布式事务。把repo clone下来之后发现包含例子只有2000多行代码，一下子震惊了。但是实际的实现还是比较简单地，当然也肯定会带来一些问题。
• 实际上上面这些API会转换为一个source，一个sink，两个map和一个包含了SerialTransactor（ProcessFunction的实现）的算子。
• 在这边展示几行代码应该就能明白是如何做到的。关键在于forceNonParallel，这就让所有事情都变得明了了，事实上就是把状态全部都托管到一个并行度为1的算子上，处理的时候也是串行的，这里我才反应过来关键在于隔离级别是Serializable。这里带来的问题就是所有状态都保存在一个节点，并且不能支持水平扩展，所能支撑的吞吐量也不能通过加机器来提升。

        SingleOutputStreamOperator<Void> resultStream = input
                .process(new SerialTransactor(specs(streamLedgerSpecs), sideOutputTags))
                .name(serialTransactorName)
                .uid(serialTransactorName + "___SERIAL_TX")
                .forceNonParallel()
                .returns(Void.class);

感想

其实看到这个功能的第一感觉是很牛逼，但是仔细看过了它的实现觉得真正应用上可能会有不少问题。因为对于最重要的处理事务的那个算子来说，本质上它并不是Scalable的，没有办法横向扩展。不过从功能上来说，确实引出了一个新的发展方向，希望以后还能看到有更优的解决方案，比如针对另外两种隔离级别Read Committed和Repeatable read。