c# – 当缓冲区不包含任何项目时,为什么Rx缓冲区连续执行方法？

我有一个Rx Observable充当缓冲区.现在,当它获得10个项目时,或者在100毫秒之后,以先到者为准,它在订阅中执行该方法.

我注意到我的方法每100毫秒不断被调用,即使缓冲区中没有项目,这让我感到惊讶.如果它没有收到缓冲区中的任何项目,那么只需让我的方法立即返回就足够了,但我觉得奇怪的是它只是在背景中像这样生长.

为什么是这样？你怎么建议我最好处​​理这个？我是Rx的全新手,所以也许我正在做一些奇怪的事情.这是我的代码的简化版本：

private Subject<keyvaluePair<int,Action<MyData>>> serverRequests;

public MyBufferClass(IMyServer server,IScheduler scheduler)
{
    this.serverRequests = new Subject<keyvaluePair<int,Action<MyData>>>();

    this.serverRequests
        .Buffer(TimeSpan.FromMilliseconds(100),10,scheduler)
        .Subscribe(buffer => GetMultipleItemsFromServer(buffer));
}   

public void GetSingleItemFromServer(int id,Action<MyData> callback)
{
    this.serverRequests.OnNext(new keyvaluePair<int,Action<MyData>>(id,callback));
}

public void GetMultipleItemsFromServer(IEnumerable<keyvaluePair<int,Action<MyData>>> idsWithCallbacks)
{
    if (idsWithCallbacks.IsNullOrEmpty()) return;

    this.server.GetMultipleItems(idsWithCallbacks)
}

在我的测试中,如果我调用GetSingleItemFromServer 5次然后将我的TestScheduler推进1000毫秒,我认为GetMultipleItemsFromServer只会被调用一次,但它会被调用10次.

解决方法

在这种情况下,一个优雅的解决方案是在Buffer之后直接使用Where运算符来过滤掉任何空结果.像这样的东西：

stream
            .Buffer (...)
            .Where (x => x.Any())
            .Subscribe (x => {...},ex => {...});

至于为什么Buffer这样做,我想最好是展示一个空集合并让消费者选择如何处理它,而不是吞下它并否认这个机会.

另外,我不会在订阅块中调用您的服务器.我认为将任何异步操作作为Rx流组合本身的一部分更好,并将Subscribe操作限制为处理最终结果的任何轻量级操作,即更新UI,记录成功/失败等等.像这样：

(from request in serverRequests
            .Buffer (TimeSpan.FromMinutes (1))
            .Where (x => x.Any())
from response in Observable.Start(server.GetMultipleItems(...))
select response)
.Subscribe (x => {},ex => {});

这方面的优点包括：

– 能够在服务器调用上使用更多Rx运算符,例如Timeout(),Retry(),Catch()等.

– 能够处理Subscribe()重载中的任何管道错误

– 使用SubscribeOn()/ ObserveOn()独立调度管道和Subscribe操作.

原文地址：https://www.jb51.cc/csharp/98863.html