c# - PubSub 在 BookSleeve/Redis 中是如何工作的？

Question

我想知道使用 BookSleeve 发布和订阅 channel 的最佳方法是什么。我目前实现了几个静态方法(见下文)，让我将内容发布到特定 channel ，并将新创建的 channel 存储在 private static Dictionary<string, RedisSubscriberConnection> subscribedChannels; 中。 .

鉴于我想发布到 channel 并订阅同一应用程序中的 channel ，这是正确的方法吗(注意:我的包装器是一个静态类)。即使我想发布和订阅，创建一个 channel 就足够了吗？显然，我不会发布到同一个 channel 而不是在同一个应用程序中订阅。但我测试了它并且它有效:

 RedisClient.SubscribeToChannel("Test").Wait();
 RedisClient.Publish("Test", "Test Message");

它奏效了。

这是我的问题:

1) 设置一个专用的发布 channel 和一个专用的订阅 channel 会比同时使用一个 channel 更有效率吗？

2)“ channel ”和“模式订阅”在语义上有什么区别？我的理解是我可以通过 PatternSubscription()订阅几个“主题”在同一个 channel ，对吗？但是，如果我想为每个“主题”调用不同的回调，我必须为每个主题设置一个 channel 是否正确？这是有效的还是你会建议不要这样做？

这里是代码片段。

谢谢!!!

    public static Task<long> Publish(string channel, byte[] message)
    {
        return connection.Publish(channel, message);
    }

    public static Task SubscribeToChannel(string channelName)
    {
        string subscriptionString = ChannelSubscriptionString(channelName);

        RedisSubscriberConnection channel = connection.GetOpenSubscriberChannel();

        subscribedChannels[subscriptionString] = channel;

        return channel.PatternSubscribe(subscriptionString, OnSubscribedChannelMessage);
    }

    public static Task UnsubscribeFromChannel(string channelName)
    {
        string subscriptionString = ChannelSubscriptionString(channelName);

        if (subscribedChannels.Keys.Contains(subscriptionString))
        {
            RedisSubscriberConnection channel = subscribedChannels[subscriptionString];

            Task  task = channel.PatternUnsubscribe(subscriptionString);

            //remove channel subscription
            channel.Close(true);
            subscribedChannels.Remove(subscriptionString);

            return task;
        }
        else
        {
            return null;
        }
    }

    private static string ChannelSubscriptionString(string channelName)
    {
        return channelName + "*";
    }

Answer 1

1:您的示例中只有一个 channel ( Test )； channel 只是用于特定发布/订阅交换的名称。但是，由于 redis API 工作方式的具体情况，有必要使用 2 个连接。具有 的连接任何 订阅除了:

收听留言

管理自己的订阅 ( subscribe , psubscribe , unsubscribe , punsubscribe )

但是，我不明白这一点:

private static Dictionary<string, RedisSubscriberConnection>

除非您要满足特定于您的需求，否则您不应该需要多个订阅者连接。单个订阅者连接可以处理任意数量的订阅。快速检查 client list在我的一台服务器上，我与(在撰写本文时)23,002 个订阅有一个连接。这可能会减少，但是:它有效。

2:模式订阅支持通配符；所以而不是订阅 /topic/1 , /topic/2/等你可以订阅 /topic/* . publish实际使用的 channel 名称作为回调签名的一部分提供给接收者。

两者都可以工作。需要注意的是 publish的性能受唯一订阅总数的影响 - 但坦率地说，即使您使用 subscribe 订阅了数万个 channel ，它仍然非常快(如:0ms)。而不是 psubscribe .

但来自 publish

Time complexity: O(N+M) where N is the number of clients subscribed to the receiving channel and M is the total number of subscribed patterns (by any client).

我建议阅读 pub/sub 的 redis 文档.

编辑以下问题:

a) I assume I would have to "publish" synchronously (using Result or Wait()) if I want to guarantee the order of sending items from the same publisher is preserved when receiving items, correct?

这根本没有任何区别；既然你提到 Result/ Wait() ，我假设您在谈论 BookSleeve - 在这种情况下，多路复用器已经保留了命令顺序。 Redis 本身是单线程的，并且总是按顺序处理单个连接上的命令。但是:订阅者上的回调可以异步执行，并且可以(单独)传递给工作线程。我目前正在调查是否可以从 RedisSubscriberConnection 强制执行此操作.

更新:从 1.3.22 开始，您可以设置 CompletionMode至 PreserveOrder - 那么所有回调将按顺序完成，而不是同时完成。

b) after making adjustments according to your suggestions I get a great performance when publishing few items regardless of the size of the payload. However, when sending 100,000 or more items by the same publisher performance drops rapidly (down to 7-8 seconds just to send from my machine).

首先，那个时间听起来很长 - 我在本地测试(对于 100,000 个出版物，包括等待所有出版物的响应)1766 毫秒(本地)或 1219 毫秒(远程)(这可能听起来违反直觉，但我的“本地”不是) t 运行相同版本的 redis；我的“远程”在 Centos 上是 2.6.12；我的“本地”是
Windows 上的 2.6.8-pre2)。

我无法使您的实际服务器更快或加快网络速度，但是:如果这是数据包碎片，我添加了(只为您) SuspendFlush()/ ResumeFlush()一对。这将禁用即时刷新(即当发送队列为空时；其他类型的刷新仍然发生)；你可能会发现这有帮助:

conn.SuspendFlush();
try {
    // start lots of operations...
} finally {
    conn.ResumeFlush();
}

请注意，您不应该 Wait直到您恢复，因为直到您调用 ResumeFlush()发送缓冲区中可能还有一些操作。一切就绪后，我得到(对于 100,000 次操作):

local: 1766ms (eager-flush) vs 1554ms (suspend-flush)
remote: 1219ms (eager-flush) vs 796ms (suspend-flush)

如您所见，它对远程服务器的帮助更大，因为它将通过网络发送更少的数据包。

I cannot use transactions because later on the to-be-published items are not all available at once. Is there a way to optimize with that knowledge in mind?

我认为上面已经解决了这个问题 - 但请注意最近 CreateBatch也被添加了。批处理的操作很像事务 - 只是:没有事务。同样，它是另一种减少数据包碎片的机制。在您的特定情况下，我怀疑暂停/恢复(刷新时)是您最好的选择。

Do you recommend having one general RedisConnection and one RedisSubscriberConnection or any other configuration to have such wrapper perform desired functions?

只要您不执行阻塞操作( blpop 、 brpop 、 brpoplpush 等)，或将过大的 BLOB 放在线路上(可能会在清除时延迟其他操作)，那么通常每种类型的单个连接效果很好。但是 YMMV 取决于您的确切使用要求。