Redis - 简单队列读取器/写入器的正确方法 - StackExchange.Redis答案

【问题标题】：Redis - correct approach for simple queue reader/writer - StackExchange.RedisRedis - 简单队列读取器/写入器的正确方法 - StackExchange.Redis
【发布时间】：2017-09-26 16:04:01
【问题描述】：

我希望使用 StackExchange.Redis 实现一个简单的分布式工作队列系统。

我理解没有BLPOP 等的原因，但就目前而言，我正在使用的界面是基于重复的TryRead 调用超时。

我对以下内容犹豫不决，因为我在处理程序中取消订阅，并设置一个标志来取消超时。有没有可能错过一些东西？是否有不同的方法来实现这一目标？

    public string TryRead(string queueName, TimeSpan timeout)
    {
        string result = null;

        var chanName = $"qnot_{queueName}";
        var done = new ManualResetEvent(false);

        void Handler(RedisChannel chan, RedisValue val)
        {
            _sub.Unsubscribe(chanName, Handler);
            result = _database.ListRightPop($"qdata_{queueName}");
            done.Set();
        }

        _sub.Subscribe(chanName, Handler);
        done.WaitOne(timeout);

        return result;
    }

    public void Write(string queueName, string text)
    {
        _database.ListLeftPush($"qdata_{queueName}", text);
        _sub.Publish($"qnot_{queueName}", "");
    }

如果队列中有现有项目（并且没有添加任何新项目），上述版本将始终超时并返回null。下面的版本现在首先检查现有数据，这是可行的。但它有一个错误，一个竞争条件：如果第一次读取检查结果为负，则推送某些内容并发送通知，然后我们订阅并等待超时。

    public string TryRead(string queueName, TimeSpan timeout)
    {
        var dataName = $"qdata_{queueName}";

        var result = (string)_database.ListRightPop(dataName);
        if (result != null)
        {
            return result;
        }

        var chanName = $"qnot_{queueName}";
        var done = new ManualResetEvent(false);

        void Handler(RedisChannel chan, RedisValue val)
        {
            _sub.Unsubscribe(chanName, Handler);
            result = _database.ListRightPop(dataName);
            done.Set();
        }

        _sub.Subscribe(chanName, Handler);
        done.WaitOne(timeout);

        return result;
    }

我可以循环执行RPOPs，但这似乎很糟糕。其他人做过类似的事情吗？

【问题讨论】：

我很困惑为什么你在处理时取消订阅 - 当然读者应该本质上是：阅读所有内容？不止一个？
我正在将其改装到当前使用 Azure 和 AWS 队列的系统中，为了保持一致，我希望为 TryRead 坚持相同的签名。它是独立的，只是阻塞直到超时@MarcGravell

标签： redis stackexchange.redis

【解决方案1】：

我最终得到了这个，它有效，但我仍然欢迎其他可行方法的答案：

    public string TryRead(string queueName, TimeSpan timeout)
    {
        var timer = Stopwatch.StartNew();
        var dataName = $"{_keyPrefix}qdata_{queueName}";
        var chanName = $"{_keyPrefix}qnot_{queueName}";
        var done = new AutoResetEvent(false);
        string result;

        // subscribe - sets the 'done' flag when a new item is pushed
        void Handler(RedisChannel chan, RedisValue val)
        {
            done.Set();
        }

        _sub.Subscribe(chanName, Handler);

        do
        {
            // try to read right away (before waiting), in case there was data already there
            result = _database.ListRightPop(dataName);
            if (result != null)
            {
                continue;
            }

            // there wasn't an item right away, so wait for the timeout to expire
            // or the subscription to be fired.  if it fired, try the read again
            var remainingTime = timeout - timer.Elapsed;
            if (remainingTime.TotalMilliseconds <= 1.0)
            {
                break;
            }
            if (done.WaitOne(remainingTime))
            {
                result = _database.ListRightPop(dataName);
            }
        } while (result == null && timer.Elapsed < timeout);

        _sub.Unsubscribe(chanName, Handler);

        return result;
    }

编辑：更新为 w/AutoResetEvent 并从处理程序中删除了 Unsubscribe。请注意那些发现这一点的人，这对我来说似乎可以替代单个阻塞读取，但这不是推荐的方法。我之所以使用它，是因为我希望与其他队列实现保持一致，并且正在使用这个特定的 TryRead 签名。

【讨论】：