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

Question

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

如果我想在同一个应用程序中发布到通道并订阅通道，这是正确的方法吗(注意:我的包装器是一个静态类)。即使我想发布和订阅，创建一个频道就足够了吗？显然，我不会发布到与在同一应用程序中订阅相同的渠道。但我测试了它，它起作用了：

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

而且它起作用了。

以下是我的问题：

1)设置一个专用的发布通道和一个专用的订阅通道是否比同时使用一个通道更有效率？

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

下面是代码片段。

谢谢！

    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:在您的示例中只有一个通道(Test)；通道只是用于特定发布/订阅交换的名称。然而，由于redis API如何工作的具体情况，使用2个连接是必要的。具有订阅的连接不能执行任何其他操作，但以下操作除外：

• 监听自己的订阅(subscribe、psubscribe、unsubscribe、punsubscribe)

然而，我不明白这个：

private static Dictionary<string, RedisSubscriberConnection>

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

2:模式订阅支持通配符；因此，您可以订阅/topic/*，而不是订阅/topic/1、/topic/2/等。publish使用的实际通道的名称将作为回调签名的一部分提供给接收方。

两者都可以工作。应该注意的是，publish的性能受到唯一订阅总数的影响-但坦率地说，即使你有成千上万的订阅频道使用subscribe而不是psubscribe，它的速度仍然非常快(如: 0ms)。

而是来自publish

时间复杂度: O(N+M)，其中N是订阅接收信道的客户端的数量，M是(任何客户端)订阅的模式的总数。

我推荐阅读pub/sub的redis文档。

对后续问题进行编辑：

a)如果我想要保证从同一发布者发送项目的顺序在接收项目时保持不变，我假设我必须同步“发布”(使用Result或Wait())，对吗？

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

更新:从1.3.22开始，您可以将CompletionMode设置为PreserveOrder -然后所有回调将按顺序完成，而不是并发完成。

b)根据您的建议进行调整后，无论有效负载的大小如何，在发布少量项目时，我都能获得很好的性能。然而，当同一发布者发送100,000或更多的项目时，性能会迅速下降(仅从我的机器发送就会下降到7-8秒)。

首先，这个时间听起来像是我在本地得到的1766ms (本地)或1219ms (远程)(这听起来可能有悖常理，但我的“本地”运行的是不同版本的redis；我的“远程”在Centos上是2.6.12；我的“本地”是Windows上的2.6.8-pre2 )。

我不能让您的实际服务器更快，也不能加快网络的速度，但是:如果这是数据包碎片，我已经(仅为您)添加了一对SuspendFlush() / ResumeFlush()。这会禁用紧急刷新(例如，当发送队列为空时；其他类型的刷新仍会发生)；您可能会发现这会有所帮助：

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

请注意，在恢复之前不应该发送，因为在调用ResumeFlush()之前，发送缓冲区中可能仍有一些操作。有了这些，我得到(对于100,000个操作)：

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

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

我不能使用事务，因为以后要发布的项目并不是一次都可用。有没有办法在考虑到这些知识的情况下进行优化？

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

您是否建议使用一个通用RedisConnection和一个RedisSubscriberConnection或任何其他配置来使这些包装器执行所需的功能？

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