如何避免在Cassandra中使用轻量级事务(CAS)时丢失写操作？

Question

我正在Cassandra上做一些测试，看看我们是否可以将它用于支持乐观并发的可伸缩的键值存储。

由于键值存储只需要一个表，并且每个项都由key访问，因此轻量级事务似乎可以轻松地为我们的问题提供技术基础。

但是，在运行执行多个并发更新的测试。时(并在检测到并发性的情况下重试)，将写入。

测试创建了一个表：

CREATE TABLE objects (key text, version int, PRIMARY KEY(key));

并使用以下方法插入多个键：

INSERT INTO objects (key, version) VALUES (?, 0) IF NOT EXISTS;

然后，使用CAS操作对这些项的版本进行多次递增：

-- client retrieves the current version
SELECT version FROM objects WHERE key = ?;

-- and updates the item using the retrieved version as version check
UPDATE objects SET version = ? WHERE key = ? IF version = ?;

对于更新，客户端代码实际上如下所示：

private async Task<bool> CompareAndSet(string key, int currrentCount, PreparedStatement updateStatement)
{
    // increment the version
    IStatement statement = updateStatement.Bind(currrentCount + 1, key, currrentCount);

    // execute the statement
    RowSet result = await Session.ExecuteAsync(statement);

    // check the result
    Row row = result.GetRows().SingleOrDefault();

    if (row == null)
        throw new Exception("No row in update result.");

    // check if the CAS operation was applied or not
    return row.GetValue<bool>("[applied]");
}

如您所见，由于并发性，无法应用CAS操作。因此，将重新尝试此操作，直到成功为止。写超时异常也会被处理。这里解释了处理写超时异常的理由。

private async Task Update(string key, PreparedStatement selectStatement, PreparedStatement updateStatement)
{
    bool done = false;

    // try update (increase version) until it succeeds
    while (!done)
    {
        // get current version                
        TestItem item = null;

        while (item == null)
            item = await GetItem(key, selectStatement);

        try
        {
            // update version using lightweight transaction 
            done = await CompareAndSet(key, item.Version, updateStatement);

            // lightweight transaction (CAS) failed, because compare failed --> simply not updated
            if (!done)
                Interlocked.Increment(ref abortedUpdates);
        }
        catch (WriteTimeoutException wte)
        {
            // partial write timeout (some have been updated, so all must be eventually updated, because it is a CAS operation)
            if (wte.ReceivedAcknowledgements > 0)
            {
                Interlocked.Increment(ref partialWriteTimeouts);
                done = true;
            }
            else
                // complete write timeout --> unsure about this one...
                Interlocked.Increment(ref totalWriteTimeouts);
        }
    }
}

下面是测试的输出，该测试使用100个项，并将每个项更新10次：

Running test with 100 items and 10 updates per item.

Number of updates: 1000
Number of aborted updates due to concurrency: 3485
Number of total write timeouts: 18
Number of partial write timeouts: 162

LOST WRITES: 94 (or 9,40%)

Results: 

Updates | Item count
     10 |         35
      9 |         43
      8 |         17
      7 |          3
      6 |          2

Xunit.Sdk.EqualExceptionAssert.Equal() Failure
Expected: 0
Actual:   94

如您所见，这是一个高度并发的测试(请参见必须重新尝试更新的中止操作的数量)。但是，坏消息是我们正在失去写。客户机认为应该执行1000个更新，但是在这个例子中94次写丢失了。

丢失写操作的次数是写超时次数的数量级。所以，它们似乎是有联系的。问题是：

• 我们需要更好地处理超时异常吗？
• 在Cassandra上执行CAS操作时，有没有避免丢失写的方法？

Answer 1

WriteTimeoutException表明卡桑德拉未能及时执行该操作。使用您的测试，您将Cassandra置于重载下，任何操作都可能在超时异常情况下失败。因此，您需要做的是重新做您的操作，并通过反复尝试从问题中恢复。它类似于SQLTimeoutException。你也需要抵抗它。