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 запись устанавливается

Пожалуйста, дайте мне знать, что исправление мне нужно сделать

Update: Моя последняя логика заключается в следующем

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

Действительно, ваш код небезопасен вокруг границы опрокидывания, потому что вы делаете «get», (латентность и мышление), «устанавливаете» - не проверяя, что условия в вашем «get» все еще применяются. Если сервер занят вокруг элемента 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-транзакции и ограничений, чтобы сделать вашу логику параллелизма -safe
2. переписать логику как сценарий Lua через ScriptEvaluate

Теперь Redis сделки (за вариант 1) are hard. Лично я бы использовал «2» - в дополнение к простому кодированию и отладке, это означает, что у вас есть только 1 раунд и работа, в отличие от «get, watch, get, multi, incr/set, exec/discard "и цикл" retry from start "для учета сценария прерывания. Я могу попытаться написать это как 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 - таким образом, при изменении значения, вы будете получать уведомления