以有限订阅者同时订阅可观察集合的简单方法

Question

我一直在尝试使用 Rx 和可观察集合来实现一个简单的生产者-消费者模式。我还需要能够轻松地限制订阅者的数量。我在并行扩展中看到了很多对 LimitedConcurrencyLevelTaskScheduler 的引用，但我似乎无法让它使用多线程。

我觉得我在做一些愚蠢的事情，所以我希望有人能解释一下。在下面的单元测试中，我希望使用多个 (2) 线程来使用阻塞集合中的字符串。我做错了什么？

[TestClass]
public class LimitedConcurrencyLevelTaskSchedulerTestscs
{
    private ConcurrentBag<string> _testStrings = new ConcurrentBag<string>();
    ConcurrentBag<int> _threadIds= new ConcurrentBag<int>();

    [TestMethod]
    public void WhenConsumingFromBlockingCollection_GivenLimitOfTwoThreads_TwoThreadsAreUsed()
    {

        // Setup the command queue for processing combinations
        var commandQueue = new BlockingCollection<string>();

        var taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(2));
        var scheduler = new TaskPoolScheduler(taskFactory);

        commandQueue.GetConsumingEnumerable()
            .ToObservable(scheduler)
            .Subscribe(Go, ex => { throw ex; });

        var iterationCount = 100;
        for (int i = 0; i < iterationCount; i++)
        {
            commandQueue.Add(string.Format("string {0}", i));
        }
        commandQueue.CompleteAdding();

        while (!commandQueue.IsCompleted)
        {
            Thread.Sleep(100);
        }

        Assert.AreEqual(iterationCount, _testStrings.Count);
        Assert.AreEqual(2, _threadIds.Distinct().Count());
    }

    private void Go(string testString)
    {
        _testStrings.Add(testString);
        _threadIds.Add(Thread.CurrentThread.ManagedThreadId);
    }
}

Answer 1

我发现通过如下修改订阅，我可以添加 5 个订阅者，但只有两个线程会处理集合的内容，所以这符合我的目的。

        for(int i = 0; i < 5; i++)
            observable.Subscribe(Go, ex => { throw ex; });

我很想知道是否有更好或更优雅的方法来实现这一点！

Answer 2

每个人似乎都经历了与 Rx 相同的学习曲线。需要理解的是，Rx 不会进行并行处理，除非您明确地进行强制并行的查询。调度器不引入并行性。

Rx 有一个行为契约，它说零个或多个值是连续产生的（不管可能使用多少线程），一个接一个，没有重叠，最后跟一个可选的单一错误或一个完整的消息，然后没有别的。

这通常写成OnNext*(OnError|OnCompleted)。

调度程序所做的就是定义规则以确定在哪个线程上处理新值如果调度程序没有正在为当前可观察对象处理的待定值。

现在获取您的代码：

var taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(2));
var scheduler = new TaskPoolScheduler(taskFactory);

这表示调度程序将为两个线程之一的订阅设置运行值。但这并不意味着它会对产生的每个价值都这样做。请记住，由于值是连续产生的，一个接一个，因此最好重新使用现有线程，而不是去创建一个新线程的高成本。所以 Rx 所做的是重新使用现有线程 if 在当前值完成处理之前，调度程序上安排了一个新值。

这是关键 - 如果在现有值的处理完成之前安排了新值，它会重新使用线程。

所以你的代码是这样做的：

commandQueue.GetConsumingEnumerable()
    .ToObservable(scheduler)
    .Subscribe(Go, ex => { throw ex; });

这意味着调度程序只会在第一个值出现时创建一个线程。但是当昂贵的线程创建操作完成时，将值添加到 commandQueue 的代码也已完成，因此它已将它们全部排队，因此它可以更有效地使用单个线程而不是创建昂贵的第二个线程.

为避免这种情况，您需要构造查询以引入并行性。

方法如下：

public void WhenConsumingFromBlockingCollection_GivenLimitOfTwoThreads_TwoThreadsAreUsed()
{
    var taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(2));
    var scheduler = new TaskPoolScheduler(taskFactory);

    var iterationCount = 100;

    Observable
        .Range(0, iterationCount)
        .SelectMany(n => Observable.Start(() => n.ToString(), scheduler)
        .Do(x => Go(x)))
        .Wait();

    (iterationCount ==  _testStrings.Count).Dump();
    (2 == _threadIds.Distinct().Count()).Dump();
}

现在，我使用了 Do(...)/.Wait() 组合来为您提供相当于阻塞 .Subscribe(...) 方法的方法。

这个结果是你的断言都返回 true。

以有限订阅者同时订阅可观察集合的简单方法

Simple way to concurrently subscribe to observable collection with limited subscribers

c#

concurrency

system.reactive

observablecollection

parallel-extensions