带锁的 Redis 分布式增量

Question

我需要生成一个计数器，该计数器将发送到一些 api 调用。我的应用程序在多个节点上运行，所以我想如何生成唯一计数器。 我试过下面的代码

public static long GetTransactionCountForUser(int telcoId)
{
    long valreturn = 0;
    string key = "TelcoId:" + telcoId + ":Sequence";
    if (Muxer != null && Muxer.IsConnected && (Muxer.GetDatabase()) != null)
    {
        IDatabase db = Muxer.GetDatabase();
        var val = db.StringGet(key);
        int maxVal = 999;
        if (Convert.ToInt32(val) < maxVal)
        {
            valreturn = db.StringIncrement(key);
        }
        else
        {
            bool isdone = db.StringSet(key, valreturn);
            //db.SetAdd(key,new RedisValue) .StringIncrement(key, Convert.ToDouble(val))
        }
    }
    return valreturn;
}

并通过 Task Parallel 库对其进行运行测试。当我有边界值时，我看到的是设置了多次 0 条目

请让我知道我需要做什么更正

更新： 我的最终逻辑如下

public static long GetSequenceNumberForTelcoApiCallViaLuaScript(int telcoId)
{
    long valreturn = 0;
    long maxIncrement = 9999;//todo via configuration
    if (true)//todo via configuration
    {
        IDatabase db;
        string key = "TelcoId:" + telcoId + ":SequenceNumber";
        if (Muxer != null && Muxer.IsConnected && (db = Muxer.GetDatabase()) != null)
        {
            valreturn = (long)db.ScriptEvaluate(@"
                local result = redis.call('incr', KEYS[1])
                if result > tonumber(ARGV[1]) then
                result = 1
                redis.call('set', KEYS[1], result)
                end
                return result", new RedisKey[] { key }, flags: CommandFlags.HighPriority, values: new RedisValue[] { maxIncrement });
        }
    }
    return valreturn;
}

Answer 1

确实，您的代码在翻转边界附近并不安全，因为您正在执行“获取”、（延迟和思考）、“设置” - 没有检查“获取”中的条件是否仍然适用。如果服务器忙于第 1000 项，则可能会得到各种疯狂的输出，包括：

1
2
...
999
1000 // when "get" returns 998, so you do an incr
1001 // ditto
1002 // ditto
0 // when "get" returns 999 or above, so you do a set
0 // ditto
0 // ditto
1

选项：

1. 使用事务和约束 API 使您的逻辑并发安全
2. 通过ScriptEvaluate将您的逻辑重写为Lua脚本

现在，redis 事务（根据选项 1）are hard。就个人而言，我会使用“2”——除了更简单的编码和调试之外，这意味着你只有 1 次往返和操作，而不是“get、watch、get、multi、incr/set、exec/丢弃”和“从开始重试”循环来说明中止情况。如果你愿意，我可以试着把它写成 Lua - 它应该是 4 行左右。

---

这是 Lua 的实现：

string key = ...
for(int i = 0; i < 2000; i++) // just a test loop for me; you'd only do it once etc
{
    int result = (int) db.ScriptEvaluate(@"
local result = redis.call('incr', KEYS[1])
if result > 999 then
    result = 0
    redis.call('set', KEYS[1], result)
end
return result", new RedisKey[] { key });
    Console.WriteLine(result);
}

注意：如果你需要参数化最大值，你会使用：

if result > tonumber(ARGV[1]) then

和：

int result = (int)db.ScriptEvaluate(...,
    new RedisKey[] { key }, new RedisValue[] { max });

（所以ARGV[1] 取值来自max）

有必要了解eval/evalsha（这是ScriptEvaluate 调用的）不与其他服务器请求竞争，因此incr 和可能set。这意味着我们不需要复杂的watch 等逻辑。

通过事务/约束 API 也是一样的（我认为！）：

static int IncrementAndLoopToZero(IDatabase db, RedisKey key, int max)
{
    int result;
    bool success;
    do
    {
        RedisValue current = db.StringGet(key);
        var tran = db.CreateTransaction();
        // assert hasn't changed - note this handles "not exists" correctly
        tran.AddCondition(Condition.StringEqual(key, current));
        if(((int)current) > max)
        {
            result = 0;
            tran.StringSetAsync(key, result, flags: CommandFlags.FireAndForget);
        }
        else
        {
            result = ((int)current) + 1;
            tran.StringIncrementAsync(key, flags: CommandFlags.FireAndForget);
        }
        success = tran.Execute(); // if assertion fails, returns false and aborts
    } while (!success); // and if it aborts, we need to redo
    return result;
}

很复杂，嗯？那么简单的成功案例是：

GET {key}    # get the current value
WATCH {key}  # assertion stating that {key} should be guarded
GET {key}    # used by the assertion to check the value
MULTI        # begin a block
INCR {key}   # increment {key}
EXEC         # execute the block *if WATCH is happy*

这是...相当多的工作，并且涉及多路复用器上的管道停顿。更复杂的情况（断言失败、监视失败、环绕）的输出会略有不同，但应该可以工作。

Answer 2

您可以使用WATCH command - 这样，如果值发生变化，您会收到通知