键值存储中的原子事务

Question

请原谅我在术语上的任何错误.特别是,我使用的是关系数据库术语.

有许多持久的键值存储,包括CouchDB和Cassandra,以及许多其他项目.

反对它们的典型论点是它们通常不允许跨多个行或表的原子事务.我想知道是否有一般方法可以解决这个问题.

以一组银行账户的情况为例.我们如何将钱从一个银行账户转移到另一个银行账户？如果每个银行帐户都是一行,我们希望将两行更新为同一事务的一部分,从而减少一个值并增加另一个值.

一种显而易见的方法是使用一个描述事务的单独表.然后,将钱从一个银行账户转移到另一个银行账户只需在该表中插入一个新行即可.我们不存储两个银行账户中任何一个的当前余额,而是依赖于汇总交易表中的所有相应行.然而,很容易想象这将是太多的工作; 银行每天可能有数百万笔交易,而个人银行账户可能很快就会有数千笔与之相关的"交易".

如果基础数据自上次抓取以来发生了变化,那么一些(全部？)键值存储将"回滚"一个动作.可能这可能用于模拟原子事务,然后,您可以指示特定字段被锁定.这种方法存在一些明显的问题.

还有其他想法吗？我的方法完全有可能是错误的,我还没有围绕新的思维方式包围我的大脑.

Answer 1

如果以您的示例为例,您希望以原子方式更新单个文档中的值(关系术语中的行),则可以在CouchDB中执行此操作.如果其他竞争客户端在您阅读之后更新了同一文档,则在尝试提交更改时会出现冲突错误.然后,您必须读取新值,更新并重新尝试提交.有一个不确定的(如果有很多争用可能是无限的)你可能需要重复这个过程的次数,但是如果你的提交成功,你可以保证在数据库中有一个原子更新的平衡文件.

如果您需要更新两个余额(即从一个帐户转移到另一个帐户),那么您需要使用单独的交易文档(实际上是行是交易的另一个表)来存储金额和两个帐户(进出) .顺便说一下,这是一种常见的簿记练习.由于CouchDB仅根据需要计算视图,因此从列出该帐户的事务计算帐户中的当前金额实际上仍然非常有效.在CouchDB中,您将使用一个映射函数,该函数将帐号作为键和事务量发出(对于传入是正数,对于传出是负数).您的reduce函数将简单地对每个键的值求和,发出相同的键和总和.然后,您可以使用group = True的视图来获取帐户余额,并按帐号键入.

Answer 2

CouchDB 不适合事务系统，因为它不支持锁定和原子操作。

为了完成银行转账，您必须做一些事情：

1. 验证交易，确保源账户中有足够的资金，两个账户都是开放的，没有被锁定，信誉良好，等等
2. 减少源账户余额
3. 增加目标账户的余额

如果在任何这些步骤之间更改帐户的余额或状态，则交易在提交后可能会失效，这在此类系统中是一个大问题。

即使您使用上面建议的方法插入“转移”记录并使用 map/reduce 视图来计算最终帐户余额，您也无法确保不会透支源帐户，因为仍然存在检查源帐户余额和插入交易之间的竞争条件，在检查余额后可以同时添加两个交易。

所以......这是工作的错误工具。CouchDB 可能擅长很多东西，但这是它真正做不到的。

编辑：可能值得注意的是，现实世界中的实际银行使用最终一致性。如果您透支银行账户的时间足够长，您会收到透支费用。如果您非常优秀，您甚至可以几乎同时从两个不同的 ATM 机取款并透支您的帐户，因为存在检查余额、发放资金和记录交易的竞争条件。当您将支票存入您的帐户时，它们会增加余额，但实际上会在一段时间内保留这些资金，“以防万一”源帐户确实没有足够的钱。

Answer 3

提供一个具体的例子（因为网上非常缺乏正确的例子）：这里是如何在 CouchDB 中实现“原子银行余额转账”（主要从我关于同一主题的博客文章中复制而来：http://blog.codekills .net/2014/03/13/atomic-bank-balance-transfer-with-couchdb/）

首先，简要回顾一下问题：如何设计允许在账户之间转移资金的银行系统，以便不存在可能留下无效或无意义余额的竞争条件？

这个问题有几个部分：

第一：事务日志。不是将账户余额存储在单个记录或文档中——{"account": "Dave", "balance": 100}账户余额是通过总结该账户的所有贷方和借方来计算的。这些贷方和借方存储在交易日志中，可能如下所示：

{"from": "Dave", "to": "Alex", "amount": 50}
{"from": "Alex", "to": "Jane", "amount": 25}

用于计算余额的 CouchDB map-reduce 函数可能如下所示：

POST /transactions/balances
{
    "map": function(txn) {
        emit(txn.from, txn.amount * -1);
        emit(txn.to, txn.amount);
    },
    "reduce": function(keys, values) {
        return sum(values);
    }
}

为完整起见，以下是余额列表：

GET /transactions/balances
{
    "rows": [
        {
            "key" : "Alex",
            "value" : 25
        },
        {
            "key" : "Dave",
            "value" : -50
        },
        {
            "key" : "Jane",
            "value" : 25
        }
    ],
    ...
}

但这留下了一个显而易见的问题：如何处理错误？如果有人试图进行大于其余额的转账，会发生什么？

对于 CouchDB（和类似的数据库），这种业务逻辑和错误处理必须在应用程序级别实现。天真地，这样的函数可能看起来像这样：

def transfer(from_acct, to_acct, amount):
    txn_id = db.post("transactions", {"from": from_acct, "to": to_acct, "amount": amount})
    if db.get("transactions/balances") < 0:
        db.delete("transactions/" + txn_id)
        raise InsufficientFunds()

但是请注意，如果应用程序在插入交易和检查更新的余额之间崩溃，则数据库将处于不一致的状态：发送方可能会留下负余额，而接收方可能会收到以前不存在的钱：

// Initial balances: Alex: 25, Jane: 25
db.post("transactions", {"from": "Alex", "To": "Jane", "amount": 50}
// Current balances: Alex: -25, Jane: 75

如何解决这个问题？

为了确保系统永远不会处于不一致状态，需要向每个事务添加两条信息：

1. 创建交易的时间（以确保交易有严格的总排序），以及

2. 状态——交易是否成功。

还需要有两个视图——一个返回账户的可用余额（即所有“成功”交易的总和），另一个返回最旧的“待处理”交易：

POST /transactions/balance-available
{
    "map": function(txn) {
        if (txn.status == "successful") {
            emit(txn.from, txn.amount * -1);
            emit(txn.to, txn.amount);
        }
    },
    "reduce": function(keys, values) {
        return sum(values);
    }
}

POST /transactions/oldest-pending
{
    "map": function(txn) {
        if (txn.status == "pending") {
            emit(txn._id, txn);
        }
    },
    "reduce": function(keys, values) {
        var oldest = values[0];
        values.forEach(function(txn) {
            if (txn.timestamp < oldest) {
                oldest = txn;
            }
        });
        return oldest;
    }

}

传输列表现在可能如下所示：

{"from": "Alex", "to": "Dave", "amount": 100, "timestamp": 50, "status": "successful"}
{"from": "Dave", "to": "Jane", "amount": 200, "timestamp": 60, "status": "pending"}

接下来，应用程序需要有一个函数，它可以通过检查每个挂起的交易来解决交易，以验证它是否有效，然后将其状态从“挂起”更新为“成功”或“拒绝”：

def resolve_transactions(target_timestamp):
    """ Resolves all transactions up to and including the transaction
        with timestamp `target_timestamp`. """
    while True:
        # Get the oldest transaction which is still pending
        txn = db.get("transactions/oldest-pending")
        if txn.timestamp > target_timestamp:
            # Stop once all of the transactions up until the one we're
            # interested in have been resolved.
            break

        # Then check to see if that transaction is valid
        if db.get("transactions/available-balance", id=txn.from) >= txn.amount:
            status = "successful"
        else:
            status = "rejected"

        # Then update the status of that transaction. Note that CouchDB
        # will check the "_rev" field, only performing the update if the
        # transaction hasn't already been updated.
        txn.status = status
        couch.put(txn)

最后，正确执行传输的应用程序代码：

def transfer(from_acct, to_acct, amount):
    timestamp = time.time()
    txn = db.post("transactions", {
        "from": from_acct,
        "to": to_acct,
        "amount": amount,
        "status": "pending",
        "timestamp": timestamp,
    })
    resolve_transactions(timestamp)
    txn = couch.get("transactions/" + txn._id)
    if txn_status == "rejected":
        raise InsufficientFunds()

一些注意事项：

• 为简洁起见，此特定实现假设 CouchDB 的 map-reduce 具有一定的原子性。更新代码使其不依赖于该假设作为练习留给读者。

• 未考虑主/主复制或 CouchDB 的文档同步。主/主复制和同步使这个问题变得更加困难。

• 在实际系统中，使用time()可能会导致冲突，因此使用具有更多熵的东西可能是个好主意；也许"%s-%s" %(time(), uuid())，或者_id在订购中使用文档。包括时间并不是绝对必要的，但如果多个请求几乎同时进入，它有助于保持逻辑。