c# - react 性扩展 : Stop an observable from returning before the tasks it has spun off have finished?

Question

我是 Rx .NET 的新手，但我有一个我认为值得的业务场景。但是，我仍然无法理解初始设计。

问题

• 我有一大堆元素，比如说 600k。
• 我有办法从数据库中分批提取这些数据(假设一次提取 1,000 个)
• 我将对这些项目并行运行一个流程，一次最多 x 数量(假设一次 50 个)
• 当我们完成后，我需要知道这一点，因为我需要吐出一些额外的统计数据并确保长时间运行的进程返回。

这似乎是响应式扩展的理想选择——我有:

• 随着时间的推移提供列表的东西
• 我想在这些元素进来时对它们做的一系列事情
• 需要处理错误
• 需要处理完成。

我从哪里开始

这似乎我会有一个项目列表作为可观察对象，我从数据库中获取这些项目的“循环”过程会将它们“推送”到这个可观察对象中，然后对该可观察对象的订阅将接管。

我卡在哪里

• 我有点不确定语法
• 我有点不确定如何处理具有限制的 x 并行度
• 我不确定我如何真正知道何时完成。那时我从数据库中提取的循环会调用“OnComplete()”而不是“OnNext”吗？

我希望有人能帮助我从概念上分解我正在寻找的东西，这样我就可以更好地思考它。谢谢!

代码 v3 -- 好多了，但该方法仍然退出得太快。

这真的开始感觉好多了，但我知道还不太好。e

public override async Task ProcessAsync(DataLoadRequest dataLoadRequest, Func<string, Task> createTrackingPayload)
{
    _requestParameters = Deserialize<SchoolETLRequestParameters>(dataLoadRequest.DataExtractorParams);
    WireUpDependencies();

    //This is the new retriever which allows records to be "paged" (e.g. returns empty list for pageNum > 0 on the ones that don't have paging.)
    _recordsToProcessRetriever = new SettingBasedRecordsRetriever(_propertyRepository, _requestParameters.RunType, _requestParameters.ResidentialProfileIDOverrides, _processorSettings.MaxBatchesToProcess, _etlLogger);

    var query = Observable.Range(0, int.MaxValue)
        .Select(pageNum => _recordsToProcessRetriever.GetResProfIDsToProcess(pageNum, _processorSettings.BatchSize))
        .TakeWhile(resProfList => resProfList.Any())
        .SelectMany(records => records)
        .Select(resProf => Observable.Start(() => Task.Run(()=> _schoolDataProcessor.ProcessSchoolsAsync(resProf)).Result))
        .Merge(maxConcurrent: _processorSettings.ParallelProperties);

    var subscription = query.Subscribe(async trackingRequests =>
        {
            await CreateRequests(trackingRequests, createTrackingPayload);
            var numberOfAttachments = SumOfRequestType(trackingRequests, TrackingRecordRequestType.AttachSchool);
            var numberOfDetachments = SumOfRequestType(trackingRequests, TrackingRecordRequestType.DetachSchool);
            var numberOfAssignmentTypeUpdates = SumOfRequestType(trackingRequests, TrackingRecordRequestType.UpdateAssignmentType);

            _etlLogger.Info("Extractor generated {0} attachments, {1} detachments, and {2} assignment type changes.",
                numberOfAttachments, numberOfDetachments, numberOfAssignmentTypeUpdates);
        },
        () =>
        {
            _etlLogger.Info("Finished! Woohoo!");
        });
}

v3 的问题

• ProcessAsync 方法仍会在后台处理完所有项目之前完成。通常我会接受，但在我们的例子中，我使用的框架需要等到所有跟踪请求都已创建(例如，直到为每批结果调用 CreateTrackingRequests)。<

是否可以等待所有操作在此期间完成？

更新:关于问题的更多信息

在这种情况下，直到运行时我们才知道什么会产生可观察对象。该应用程序通过命令传递，相当于:

• “新记录”:命中一个返回特定存储过程结果的方法
• “特定记录”:用于测试；命中一个方法，该方法为特定给定值命中一个单独的存储过程
• “所有记录”:点击进入连续分页循环的方法，在 x 页(由设置定义)中循环遍历 600k 条记录。

前两个场景听起来我可以毫无问题地将它们直接传递到一个可观察对象中。但是，在这种情况下，最后一个似乎我必须循环遍历一组可观察对象，这不是我想要的行为(我希望所有 600k 项目最终都在一个大队列中并在 50 处处理一次)。

我希望我能有一种方法“将事情扔到队列中”，并让处理任务以 50 个为一批连续地从中拉取。

注意:所有那些调用存储过程的方法都返回完全相同的东西——IThing 的列表。 (出于必要而混淆)。

我已将所有这些存储库函数等连接到我的处理器 AS 依赖项中，因此调用 ProcessStuffForMyThing(List<IThing>)负责整个过程，并使用同一个对象并行工作(不需要每次都更新它)。

Answer 1

您的代码有许多问题需要解决。你犯的错误我见过很多次——每个人似乎都在走同一条路。这实际上归结为将您的思维方式从程序性转变为功能性。

首先，Rx 有很多旨在让您的生活更轻松的运算符。其中之一是 Observable.Using。它的工作是启动一次性资源，构建可观察对象，并在可观察对象完成时处理资源。非常适合从数据库中读取记录。

您的代码似乎有一个已经打开的数据库连接，并且您正在通过主题抽取记录。您应该避免使用外部状态(数据处理器)，并且应该避免使用主题。几乎总有一个您可以使用的可观察运算符。

您正在做的另一件您可能不应该做的事情是混合您的 monad - 或者更具体地说是可观察对象和任务。 Rx 中有一些运算符可以将任务转换为可观察对象，但它们是用来与现有代码交互的，不应该用作可观察对象中的工具。规则是尝试进入可观察对象并停留在那里，直到您准备好订阅您的数据。

我觉得您的代码有点零散，无法准确理解在哪里调用了什么，所以我写了一段通用代码，我认为它涵盖了您的需求。这是查询:

var pageSize = 4;

Func<Record, Result> process = r =>
{
    Thread.Sleep(100); // Only here to demonstrate parallelism
    return new Result(r.ID);
};

var query =
    Observable
        .Using(
            () => new DataProcessor(),
            dc =>
                Observable
                    .Range(0, int.MaxValue)
                    .Select(n => dc.GetRecords(n, pageSize))
                    .TakeWhile(rs => rs.Any())
                    .SelectMany(rs => rs)
                    .Select(r => Observable.Start(() => process(r)))
                    .Merge(maxConcurrent: 4));

var subscription =
    query
        .Subscribe(
            r => Console.WriteLine(r.ID),
            () => Console.WriteLine("Done."));

我显然在您的代码中采用了一些捷径，但本质上它们大同小异(我希望如此)。

如果您添加以下类，此代码是可运行的:

public class DataProcessor : IDisposable
{
    public DataProcessor() { Console.WriteLine("Opened."); }
    public void Dispose() { Console.WriteLine("Closed."); }
    public IEnumerable<Record> GetRecords(int page, int count)
    {
        Console.WriteLine("Reading.");
        Thread.Sleep(100);
        var records = page <= 5
            ? Enumerable
                .Range(0, count < 5 ? count : count / 2)
                .Select(x => new Record())
                .ToArray()
            : new Record[] { };
        Console.WriteLine("Read.");
        return records;
    }
}

public class Record
{
    private static int __counter = 0;
    public Record() { this.ID = __counter++; }
    public int ID { get; private set; }
}

public class Result
{
    public Result(int id) { this.ID = id; }
    public int ID { get; private set; } 
}

当我运行它时，我得到了这个结果:

Opened.
Reading.
Read.
Reading.
0
2
3
1
Read.
Reading.
7
Read.
5
6
4
Reading.
10
11
9
8
Read.
Reading.
15
12
Read.
14
Reading.
13
17
19
18
16
Read.
Reading.
21
Read.
20
22
23
Done.
Closed.

可以看到是并行处理的。您可以看到 observable 正在完成。您还可以看到数据库正在打开，然后在 observable 完成后关闭。

如果这有帮助，请告诉我。