为什么我的 Observable 中的 Buffer 没有收到事件？

Question

我尝试了几种方法，但未能调用我的订阅方法：

• 方法一：

  var buffer = new List<Kpi>();
  buffer.ToObservable().Buffer(TimeSpan.FromMinutes(1), 5).Subscribe(
  async kpis =>
  {
      await _retry.ExecuteAsync(() => Process(kpis.ToList())).ConfigureAwait(false);
  });
  

  那么buffer.Add(new Kpi()); 不会触发我的方法。

---

• 方法 2：（注意：我已经阅读了特殊方法 Empty/Never/Throw 的定义，但除此之外，我似乎无法找到一种方法来创建一个可以发出除原始数字以外的其他东西的可观察对象。）

  var buffer = Observable.Empty<Kpi>();
  buffer.Buffer(TimeSpan.FromMinutes(1), 5).Subscribe(
  async kpis =>
  {
      await _retry.ExecuteAsync(() => Process(kpis.ToList())).ConfigureAwait(false);
  });
  

  然后buffer.Publish(new Kpi()) 。再次没有任何反应

我哪里出错了？

Answer 1

在第一种情况下，在List 上调用ToObservable 不会让List 神奇地通知它的更改。 List 根本没有这个功能。

在第二种情况下，Publish 所做的事情与您的预期完全不同。

如果你想从事件中创建一个 observable，你正在寻找 Subject 类。

var buffer = new Subject<Kpi>();
buffer.Buffer(TimeSpan.FromMinutes(1), 5).Subscribe(
async kpis =>
{
    await _retry.ExecuteAsync(() => Process(kpis.ToList())).ConfigureAwait(false);
});

// notify of new item
buffer.OnNext(new Kpi());

有很多方法可以创建new observable sequence。我建议您通读一遍，看看是否更适合您。例如将事件变成可观察的。