可在 C# 中使用背压进行观察

Question

C# rx 中有处理背压的方法吗？我正在尝试从分页查询的结果中调用网络 api。这个网站 api 非常脆弱，我需要的并发调用不超过 3 个，因此，程序应该是这样的：

从 db 获取一个页面
调用网络 api 页面上每条记录最多同时调用三个
将结果保存回数据库
获取另一页并重复直到没有更多结果。

我并没有真正得到我想要的序列，基本上数据库会得到所有的记录，不管它们是否可以被处理。

我尝试了各种方法，包括调整 ObserveOn 运算符、实现信号量以及其他一些方法。我可以得到一些指导来实现这样的东西吗？

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;
using System.Threading;
using System.Threading.Tasks;
using Castle.Core.Internal;
using Xunit;
using Xunit.Abstractions;

namespace ProductValidation.CLI.Tests.Services
{
    public class Example
    {
        private readonly ITestOutputHelper output;

        public Example(ITestOutputHelper output)
        {
            this.output = output;
        }

        [Fact]
        public async Task RunsObservableToCompletion()
        {
            var repo = new Repository(output);
            var client = new ServiceClient(output);

            var results = repo.FetchRecords()
                .Select(x => client.FetchMoreInformation(x).ToObservable())
                .Merge(1)
                .Do(async x => await repo.Save(x));

            await results.LastOrDefaultAsync();
        } 
    }

    public class Repository
    {
        private readonly ITestOutputHelper output;

        public Repository(ITestOutputHelper output)
        {
            this.output = output;
        }

        public IObservable<int> FetchRecords()
        {
            return Observable.Create<int>(async (observer) =>
            {
                var page = 1;
                var products = await FetchPage(page);
                while (!products.IsNullOrEmpty())
                {
                    foreach (var product in products)
                    {
                        observer.OnNext(product);
                    }

                    page += 1;
                    products = await FetchPage(page);
                }
                observer.OnCompleted();
            })
            .ObserveOn(SynchronizationContext.Current);
        }

        private async Task<IEnumerable<int>> FetchPage(int page)
        {
            // Simulate fetching a paged query.
            await Task.Delay(500).ToObservable().ObserveOn(new TaskPoolScheduler(new TaskFactory()));
            output.WriteLine("Fetching page {0}", page);
            if (page >= 4) return Enumerable.Empty<int>();
            return Enumerable.Range(1, 3).Select(_ => page);
        }

        public async Task Save(string id)
        {
            await Task.Delay(50); //Simulates latency
        }
    }

    public class ServiceClient
    {
        private readonly ITestOutputHelper output;
        private readonly SemaphoreSlim semaphore;

        public ServiceClient(ITestOutputHelper output)
        {
            this.output = output;
            this.semaphore = new SemaphoreSlim(2);
        }

        public async Task<string> FetchMoreInformation(int id)
        {
            try
            {
                output.WriteLine("Calling the web client for {0}", id);
                await semaphore.WaitAsync(); // Protection for the webapi not sending too many calls
                await Task.Delay(1000); //Simulates latency
                return id.ToString();
            }
            finally
            {
                semaphore.Release();
            }
        }
    }
}

Answer 1

Rx 不支持背压，因此没有简单的方法以与处理记录相同的速度从数据库中获取记录。也许您可以使用 Subject<Unit> 作为信号机制，每次处理记录时推送一个值，并设计一种方法在生产站点使用这些信号在收到信号时从数据库中获取新记录.但这将是一个混乱且惯用的解决方案。 TPL Dataflow is a more suitable tool than the Rx for doing this kind of work. It supports natively the BoundedCapacity 配置选项。

关于您发布的代码的一些评论，与背压问题没有直接关系：

带有maxConcurrent参数的Merge运算符对内部序列的并发订阅施加了限制，但是如果内部序列已经启动并且运行.所以你必须确保内部序列是冷的，一个方便的方法是 Defer 运算符：

.Select(x => Observable.Defer(() =>
    client.FetchMoreInformation(x).ToObservable()))

将异步方法转换为延迟可观察序列的更常见方法是 FromAsync 运算符：

.Select(x => Observable.FromAsync(() => client.FetchMoreInformation(x)))

顺便说一下，Do 运算符不理解异步委托，所以不是：

.Do(async x => await repo.Save(x));

...创建 async void lambdas，最好这样做：

.Select(x => Observable.FromAsync(() => repo.Save(x)))
.Merge(1);

更新： 下面是一个如何使用 SemaphoreSlim 来在 Rx 中实现背压的例子：

const int boundedCapacity = 10;
using var semaphore = new SemaphoreSlim(boundedCapacity, boundedCapacity);

IObservable<int> results = repo
    .FetchRecords(semaphore)
    .Select(x => Observable.FromAsync(() => client.FetchMoreInformation(x)))
    .Merge(1)
    .Select(x => Observable.FromAsync(() => repo.Save(x)))
    .Merge(1)
    .Do(_ => semaphore.Release());

await results.DefaultIfEmpty();

在 FetchRecords 方法中：

//...
await semaphore.WaitAsync();
observer.OnNext(product);
//...

这是一个脆弱的解决方案，因为它依赖于通过管道传播所有元素。如果将来您决定在管道内包含过滤或节流，那么 WaitAsync 和 Release 之间的一对一关系将被违反，最有可能的结果是管道死锁。

可在 C# 中使用背压进行观察

Observable with backpressure in C#

c#

system.reactive