RX：如何等待订阅者完成？

Question

我有一个生产者，它从Rest API下载页面中的数据，以及几个处理页面的消费者（例如，将它们加载到数据库中）。

我希望生产者和消费者并行工作，这意味着生产者不应该等到一个页面被消费后再下载下一个页面。每个使用者都需要按顺序处理页面。

当下载所有页面时，主线程应该等待所有消费者完成他们的工作（因为消费可能比生产需要更长的时间）。

我目前的做法如下：

我已经创建了一个下载页面的可观察对象，它在附加消费者订阅者后立即开始。我将订阅者配置为在自己的线程上观察以进行并行执行。

C# 中的代码：

IEnumerable<Page> getPages = produce();
var observable = getPages.ToObservable().Publish();

observable
   .ObserveOn(NewThreadScheduler.Default)
   .Subscribe(page => consume1(page));

observable
   .ObserveOn(NewThreadScheduler.Default)
   .Subscribe(page => consume2(page));

observable.Connect();

这种实现的问题是，主线程可能会在所有页面都处理完毕且应用程序停止之前结束。

如何使用 RX 实现此目的？

谢啦！

编辑:

还尝试了以下方法（来自答案）：

static void Main(string[] args)
{
    var getPages = Enumerable.Range(0, 10);

    var els1 = new EventLoopScheduler();
    var els2 = new EventLoopScheduler();

    var observable =
        getPages
            .ToObservable()
            .Publish(ps =>
                Observable
                    .Merge(
                        ps.Select(p => Observable.Start(() => consume1(p), els1)),
                        ps.Select(p => Observable.Start(() => consume2(p), els2))));

    observable.Wait();
}

public static void consume1(int p)
{
    Console.WriteLine($"1:{p}");
    Thread.Sleep(200);
}

public static void consume2(int p)
{
    Console.WriteLine($"2:{p}");
    Thread.Sleep(100);
}

observable.Wait（）在基础可枚举完成产生值后立即返回。输出是：

1:0
2:0

只是为了证明，如果我们用以下命令替换 getPages：

var getPages = Enumerable.Range(0, 10)
    .Select(i =>
    {
        Console.WriteLine($"Produced {i}");
        Thread.Sleep(30);
        return i;
    });

那么输出是:

Produced 0
Produced 1
1:0
2:0
Produced 2
Produced 3
Produced 4
2:1
Produced 5
Produced 6
Produced 7
1:1
2:2
Produced 8
Produced 9

Answer 1

我想这正是你想要的:

var els1 = new EventLoopScheduler();
var els2 = new EventLoopScheduler();

var observable =
    getPages
        .ToObservable()
        .Publish(ps =>
            Observable
                .Merge(
                    ps.SelectMany(p => Observable.Start(() => consume1(p), els1)),
                    ps.SelectMany(p => Observable.Start(() => consume2(p), els2))));

我写了这个测试代码：

var getPages = Enumerable.Range(0, 10);

var els1 = new EventLoopScheduler();
var els2 = new EventLoopScheduler();

var observable =
    getPages
        .ToObservable()
        .Publish(ps =>
            Observable
                .Merge(
                    ps.SelectMany(p => Observable.Start(() => consume1(p), els1)),
                    ps.SelectMany(p => Observable.Start(() => consume2(p), els2))));

observable.Wait();

public void consume1(int p)
{
    Console.WriteLine($"1:{p}");
    Thread.Sleep(200);
}

public void consume2(int p)
{
    Console.WriteLine($"2:{p}");
    Thread.Sleep(100);
}

我得到了这样的输出:

1:0
2:0
2:1
1:1
2:2
2:3
1:2
2:4
2:5
1:3
2:6
2:7
1:4
2:8
2:9
1:5
1:6
1:7
1:8
1:9

当您完成EventLoopScheduler实例时，您应该调用。对它们进行Dispose（）以关闭线程。