c# - StackExchange.Redis 阻塞弹出设计

Question

因此，我们有一个使用 ServiceStack.Redis 的现有 Helper 库，目前正在尝试将其与 StackExchange.Redis 交换。我们使用的是 BlockingPop (BLPOP)，但由于 StackExchange.Redis 不支持它。我们实现如下

public static void Push(string Qname, string val)
{
    IDatabase db = redis.GetDatabase();
    db.ListLeftPush(Qname, val);
    ISubscriber sub = redis.GetSubscriber();
    sub.Publish(Qname + "_msg", "1");
}

和带有阻塞选项的 Pop 如下:

    public static string Pop(string Qname, 
    bool block_until_available = false,int timeout_secs=0)
{   
    IDatabase db = redis.GetDatabase();            
    var popped = db.ListRightPop(Qname);
    if (popped.IsNull)
    {
        if (block_until_available == false)
            return null;
    }
    else
        return popped;

    //wait for an item to be pushed in.
    ISubscriber sub = redis.GetSubscriber();
    AutoResetEvent autoEvent = new AutoResetEvent(false);
    string obj = null;
    Task.Run(() =>
    {
        sub.Subscribe(Qname + "_msg", (channel, message) =>
        {
            popped = db.ListRightPop(Qname);
            if (!popped.IsNull)
            {
                obj = popped;
                sub.Unsubscribe(Qname + "_msg");
                autoEvent.Set();
            }
        });
    });
    if (timeout_secs > 0)
        autoEvent.WaitOne(timeout_secs * 1000);
    else
        autoEvent.WaitOne();
    return obj;
}

你们都看到这种方法有什么明显的问题吗？

此外，我很快遇到了以下错误。我增加了 syncTimeout。希望这会解决问题？

System.TimeoutException: Timeout performing RPOP DL_PROD, 
inst: 0, mgr: ProcessReadQueue, err: never, queue: 0, qu: 0, qs: 0, qc: 0, 
wr: 0, wq: 0, in: 0, ar: 1, IOCP: (Busy=0,Free=1000,Min=8,Max=1000), 
WORKER: (Busy=2,Free=32765,Min=8,Max=32767), 
clientName: CD147RE1 at 
StackExchange.Redis.ConnectionMultiplexer.ExecuteSyncImpl[T]
(Message message, ResultProcessor`1 processor, ServerEndPoint server) 

Answer 1

RPOP 在这里超时没有具体原因，除非它与带宽相关(巨大的有效负载)。该错误看起来与 IMO 问题无关。这种方法很丑陋，但是……好吧，它可能会奏效。不过，这里没有必要使用 Task.Run。我认为，该方法的一个问题是它不能同时正确工作。它似乎取消订阅该 channel /连接的所有代表，而不仅仅是一个。自行退订是可能的，但坦率地说，我想知道只有一个自动重置事件和一个订阅是否更容易，如果一条消息在你等待时打开大门:太好了 .

---

自行取消订阅受托人的一般模式本质上是:

YourDelegateType handler = null;
handler = (args) => {
    DoTheThing();
    UnSubscribe(handler);
};
Subscribe(handler);

这利用词法捕获变量的乐趣来提供对委托(delegate)内部委托(delegate)实例的访问，这意味着委托(delegate)可以将自身传递给取消订阅方法。上面显示的模式应该适用于所有基于委托(delegate)的回调场景——包括调节事件和 SE.Redis 发布/订阅处理程序之类的东西。