ReactiveExtensions:阻止一个可观察对象在它剥离的任务完成之前返回

您的代码有许多需要修复的问题。你犯的错误我已经见过很多次了——每个人似乎都在走同样的路。这实际上可以归结为将你的思维从过程性转变为功能性。

首先，Rx有许多操作符，旨在使您的工作更轻松。其中之一是Observable.Using。它的工作是启动一个可丢弃的资源，构建一个可观察对象，并在可观察对象完成时处置该资源。非常适合从数据库中读取记录。

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

另一件你可能不应该做的事情是混合你的单子——或者更具体地说，是可观察对象和任务。在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.

您可以看到它正在并行处理。你可以看到可观察对象正在完成。你还可以看到数据库正在打开，然后在可观察对象完成后关闭。

让我知道这是否有帮助。

首先，我不建议滚动您自己的枚举转换。如果您有IEnumerable<T>，您可以使用.ToObservable()扩展，它将为您处理枚举。

其次，您应该在Subscribe方法中处理Observable的结果，现在您的方法将在枚举之后立即返回，因为您似乎没有在async方法中实际等待任何东西。如果你必须使用当前方法签名，那么你可以利用可观察对象也是可等待的。

public override async Task ProcessAsync(Request theRequest,
                             Func<string,Task> createTrackingPayload) // not my design^TM
    {
        // ...do stuff with the request, wire up some dependencies, etc.
        //End goal is to call createTrackingPayload with some things.
            await items.ToObservable()
            .Select(thing => Observable.FromAsync(async () =>
            {
                var requests = await _dataProcessor.DoSomethingAsync(thing);
                if (requests != null && requests.Any())
                {
                    var numberOfType1 = SumOfRequestType(requests, TrackingRecordRequestType.Type1);
                    var numberOfType2 = SumOfRequestType(requests, TrackingRecordRequestType.DetachSchool);
                    var numberOfType3 = SumOfRequestType(requests, TrackingRecordRequestType.UpdateAssignmentType);

                    await CreateRequests(requests, createTrackingPayload); // something that will iterate over the list and call the function we need to call.
                    return requests.Count();
                }
                return 0;
            }
            }))
        .Merge(maxConcurrent: _processorSettings.DegreeofParallelism)
        .Do(x => _logger.Info("processed {0} items.", x))
        .Aggregate(0, (acc, x) => acc + x);
    }

基本上，这里的想法是等待Observable的完成，它实际上会给你Observable完成之前的最后一个值。通过添加Do和Aggregate，您可以将日志记录逻辑移出处理逻辑。

我在这里把功劳归功于Enigmativity，因为他们的答案引导我找到了(大部分)正确的地方。

我需要的代码如下，除了序列多次求值的一个小问题。

var query = Observable.Range(0, int.MaxValue)
    .Select(pageNum =>
        {
            _etlLogger.Info("Calling GetResProfIDsToProcess with pageNum of {0}", pageNum);
            return _recordsToProcessRetriever.GetResProfIDsToProcess(pageNum, _processorSettings.BatchSize);
        })
    .TakeWhile(resProfList => resProfList.Any())
    .SelectMany(records => records.Where(x=> _determiner.ShouldProcess(x)))
    .Select(resProf => Observable.Start(async () => await _schoolDataProcessor.ProcessSchoolsAsync(resProf)))
    .Merge(maxConcurrent: _processorSettings.ParallelProperties)
    .Do(async trackingRequests =>
    {
        await CreateRequests(trackingRequests.Result, createTrackingPayload);
        var numberOfAttachments = SumOfRequestType(trackingRequests.Result, TrackingRecordRequestType.AttachSchool);
        var numberOfDetachments = SumOfRequestType(trackingRequests.Result, TrackingRecordRequestType.DetachSchool);
        var numberOfAssignmentTypeUpdates = SumOfRequestType(trackingRequests.Result,
            TrackingRecordRequestType.UpdateAssignmentType);
        _etlLogger.Info("Extractor generated {0} attachments, {1} detachments, and {2} assignment type changes.",
            numberOfAttachments, numberOfDetachments, numberOfAssignmentTypeUpdates);
    });
var subscription = query.Subscribe(
trackingRequests =>
{
    //Nothing really needs to happen here. Technically we're just doing something when it's done.
}, 
() =>
{
    _etlLogger.Info("Finished! Woohoo!");
});
await query.Wait();