带锁定的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 libray运行测试。当我有边界值时，我看到的是设置了多次0条目

请让我知道我需要做的更正

更新：我的最终逻辑如下

 public static long GetSequenceNumberForTelcoApiCallViaLuaScript(int telcoId)
    {
        long valreturn = 0;
        int 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 = (int)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. 通过Luahtml" target="_blank">脚本重写您的逻辑 ScriptEvaluate

现在，redis事务（按选项1）很困难。就我个人而言，我将使用“
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*

这是一项相当大的工作，涉及多路复用器上的流水线停顿。更复杂的情况（断言失败，监视失败，环绕）将具有稍微不同的输出，但应该可以工作。