键值存储中的原子事务

以您的示例为例,如果您希望以原子方式更新 仅有一个的 文档(关系术语中的行),您可以在CouchDB中这样做。如果其他争用客户端在您读取同一文档后更新了该文档,则在尝试提交更改时将出现冲突错误。然后必须读取新值,更新并重新尝试提交。存在不确定(如果存在 大量 您可能必须重复此过程的次数,但如果提交成功,则保证数据库中的文档具有原子更新的平衡。

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

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

为了完成银行转账,您必须做以下几件事:

1. 验证交易,确保源帐户中有足够的资金,两个帐户都已打开、未锁定且信誉良好,等等
2. 减少源帐户的余额

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

即使您使用上述方法插入“转账”记录并使用map/REDUCT视图计算最终账户余额,您无法确保不透支源帐户,因为在检查源帐户余额和插入交易(在检查余额后可以同时添加两个交易)之间仍然存在竞争条件。

所以这是做这项工作的错误工具。CouchDB可能擅长很多事情,但这是它真正做不到的。

编辑:可能值得注意的是,现实世界中的实际银行使用最终一致性。如果你透支你的银行账户足够长的时间,你会得到一笔透支费。如果你做得很好,你甚至可以在几乎同一时间从两台不同的自动取款机上取款,并透支你的账户,因为有一个种族条件来检查余额、发款和记录交易。当你把支票存入你的账户时,他们会冲破余额,但实际上会在一段时间内持有这些资金,以“以防万一”源账户没有足够的资金。

atomic bank balance transfer “在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
        }
    ],
    ...
}

但这留下了一个显而易见的问题:错误是如何处理的?如果 有人试图进行超过余额的转账?

处理必须在应用程序级别实现。天真地说,这样的功能 可能是这样的:

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. 创建事务的时间(以确保 strict total ordering 以及

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地图中的原子性减少。更新代码,使其不依赖于

• 主/主复制或CouchDB的文档同步尚未考虑在内 考虑主/主复制和同步会导致此问题 要困难得多。

• 在实际系统中,使用 time() 可能会导致碰撞,因此使用 熵多一点的东西可能是个好主意;大概 "%s-%s" %(time(), uuid()) ,或使用文档的 _id 在订单中。 包括时间并不是绝对必要的,但它有助于保持逻辑性

BerkeleyDB和LMDB都是支持ACID事务的键值存储。在BDB中,TXN是可选的,而LMDB仅以事务方式运行。

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

如果数据存储支持每键线性化、比较和交换或测试和设置操作,那么就足以实现可序列化事务。例如,此方法用于 Google's Percolator 而且 CockroachDB

在我的博客中,我创建了 step-by-step visualization of serializable cross shard client-side transactions ,描述了主要用例,并提供了指向算法变体的链接。我希望它能帮助您理解如何为您的数据存储实现它们。

• 里亚克水桶一致
• 重新思考数据库
• Etdc
• 发电机

顺便说一句,如果您对读取提交隔离级别没有意见,那么可以考虑一下 RAMP transactions 彼得·贝利斯著。它们也可以为同一组数据存储实现。