34

我将 Redis 与 StackExchange.Redis 一起使用。我有多个线程会在某个时候访问和编辑同一个键的值，所以我需要同步数据的操作。

查看可用函数，我看到有两个函数，TakeLock 和 ReleaseLock。但是，这些函数需要一个键和一个值参数，而不是预期的单个键被锁定。GitHub 上的 intellisene 文档和源代码没有解释如何使用 LockTake 和 LockRelease 函数或为键和值参数传递什么。

问：StackExchange.Redis 中 LockTake 和 LockRelease 的正确用法是什么？

我打算做的伪代码示例：

//Add Items Before Parallel Execution
redis.StringSet("myJSONKey", myJSON);

//Parallel Execution
Parallel.For(0, 100, i =>
    {
        //Some work here
        //....

        //Lock
        redis.LockTake("myJSONKey");

        //Manipulate
        var myJSONObject = redis.StringGet("myJSONKey");
        myJSONObject.Total++;
        Console.WriteLine(myJSONObject.Total);
        redis.StringSet("myJSONKey", myNewJSON);

        //Unlock
        redis.LockRelease("myJSONKey");

        //More work here
        //...
    });
4

2 回答 2

59

锁有3个部分：

  • 密钥（数据库中锁的唯一名称）
  • 值（调用者定义的标记，可用于指示谁“拥有”锁，并检查释放和扩展锁是否正确完成）
  • 持续时间（故意锁定是有限持续时间的东西）

如果没有想到其他值，则 guid 可能会生成一个合适的“值”。我们倾向于使用机器名称（如果多个进程可能在同一台机器上竞争，则使用机器名称的混合版本）。

另外，请注意，锁定是推测性的，而不是阻塞性的。您完全有可能无法获得锁，因此您可能需要对此进行测试，并可能添加一些重试逻辑。

一个典型的例子可能是：

RedisValue token = Environment.MachineName;
if(db.LockTake(key, token, duration)) {
    try {
        // you have the lock do work
    } finally {
        db.LockRelease(key, token);
    }
}

请注意，如果工作很长（特别是循环），您可能希望LockExtend在中间添加一些偶尔的调用 - 再次记住检查是否成功（以防超时）。

另请注意，所有单独的 redis 命令都是原子的，因此您无需担心两个谨慎的操作竞争。对于更复杂的多操作单元，事务和脚本是选项。

于 2014-08-05T11:53:12.507 回答
5

我的代码部分用于 lock->get->modify(if required)->unlock 带有注释的操作。

    public static T GetCachedAndModifyWithLock<T>(string key, Func<T> retrieveDataFunc, TimeSpan timeExpiration, Func<T, bool> modifyEntityFunc,
       TimeSpan? lockTimeout = null, bool isSlidingExpiration=false) where T : class
    {
        
        int lockCounter = 0;//for logging in case when too many locks per key
        Exception logException = null;

        var cache = Connection.GetDatabase();
        var lockToken = Guid.NewGuid().ToString(); //unique token for current part of code
        var lockName = key + "_lock"; //unique lock name. key-relative.
        T tResult = null;
        
        while ( lockCounter < 20)
        {
            //check for access to cache object, trying to lock it
            if (!cache.LockTake(lockName, lockToken, lockTimeout ?? TimeSpan.FromSeconds(10)))
            {
                lockCounter++;
                Thread.Sleep(100); //sleep for 100 milliseconds for next lock try. you can play with that
                continue;
            }

            try
            {
                RedisValue result = RedisValue.Null;

                if (isSlidingExpiration)
                {
                    //in case of sliding expiration - get object with expiry time
                    var exp = cache.StringGetWithExpiry(key);
                    
                    //check ttl.
                    if (exp.Expiry.HasValue && exp.Expiry.Value.TotalSeconds >= 0)
                    {
                        //get only if not expired
                        result = exp.Value;
                    }
                }
                else //in absolute expiration case simply get
                {
                    result = cache.StringGet(key);
                }

                //"REDIS_NULL" is for cases when our retrieveDataFunc function returning null (we cannot store null in redis, but can store pre-defined string :) )
                if (result.HasValue && result == "REDIS_NULL") return null;
                //in case when cache is epmty
                if (!result.HasValue)
                {
                    //retrieving data from caller function (from db from example)
                    tResult = retrieveDataFunc();

                    if (tResult != null)
                    {
                        //trying to modify that entity. if caller modifyEntityFunc returns true, it means that caller wants to resave modified entity.
                        if (modifyEntityFunc(tResult))
                        {
                            //json serialization
                            var json = JsonConvert.SerializeObject(tResult);
                            cache.StringSet(key, json, timeExpiration);
                        }
                    }
                    else
                    {
                        //save pre-defined string in case if source-value is null.
                        cache.StringSet(key, "REDIS_NULL", timeExpiration);
                    }
                }
                else
                {
                    //retrieve from cache and serialize to required object
                    tResult = JsonConvert.DeserializeObject<T>(result);
                    //trying to modify
                    if (modifyEntityFunc(tResult))
                    {
                        //and save if required
                        var json = JsonConvert.SerializeObject(tResult);
                        cache.StringSet(key, json,  timeExpiration);
                    }
                }

                //refresh exiration in case of sliding expiration flag
                if(isSlidingExpiration)
                    cache.KeyExpire(key, timeExpiration);
            }
            catch (Exception ex)
            {
                logException = ex;
            }
            finally
            {                    
                cache.LockRelease(lockName, lockToken);
            }
            break;
        }

        if (lockCounter >= 20 || logException!=null)
        {
            //log it
        }

        return tResult;
    }

和用法：

public class User
{
    public int ViewCount { get; set; }
}

var cachedAndModifiedItem = GetCachedAndModifyWithLock<User>( 
        "MyAwesomeKey", //your redis key
        () => // callback to get data from source in case if redis's store is empty
        {
            //return from db or kind of that
            return new User() { ViewCount = 0 };
        }, 
        TimeSpan.FromMinutes(10), //object expiration time to pass in Redis
        user=> //modify object callback. return true if you need to save it back to redis
        {
            if (user.ViewCount< 3)
            {
                user.ViewCount++;
                return true; //save it to cache
            }
            return false; //do not update it in cache
        },
        TimeSpan.FromSeconds(10), //lock redis timeout. if you will have race condition situation - it will be locked for 10 seconds and wait "get_from_db"/redis read/modify operations done.
        true //is expiration should be sliding.
        );

该代码可以改进（例如，您可以添加事务以减少对缓存的计数调用等），但我很高兴它对您有所帮助。

于 2015-09-29T19:42:55.353 回答