带有单个任务的异步 OnNext答案

【问题标题】：async OnNext with single task带有单个任务的异步 OnNext
【发布时间】：2020-05-13 12:28:18
【问题描述】：

我有一个每 100 毫秒获取一个 int 的主题。这个数字在每次迭代中递增。我订阅了这个主题，拿回 50 块，然后等待几秒钟。示例：

SequenceGenerator gen = new SequenceGenerator();
gen.subj.Buffer(50).Subscribe(
    async x =>
    {
        Console.WriteLine($"1: {x[x.Count - 1]}------");
        await Task.Delay(10000);
        Console.WriteLine($"1:------");
    },
    x => Console.WriteLine("EX"),
    () => Console.WriteLine("1 Done"));

SequenceGenerator 仅执行以下操作：

int i = 0;
while (true)
{
    ++i;
    subj.OnNext(i.ToString());
    await Task.Delay(1);
}

每次缓冲区被填满时，都会调用“onnext”。即使最后一个“onnext”还没有完成。有没有办法调整这个？如果最后一个“onnext”完成，我希望只调用下一个“onnext”。实际行为：

1: 50------
(wait ~0.5s)
1: 100------
...

预期行为：

1: 50------
(wait 10 seconds)
1: 100------
...

有没有办法做到这一点？谢谢大家！最好的问候。

【问题讨论】：

标签： .net system.reactive reactive subject

【解决方案1】：

好的，我想我有一个解决方案。我找到了一个提示here。 Zip 通过这种方法帮助了我很多。我需要另一个压缩到我的流中的主题。至少这是可行的：

SequenceGenerator gen = new SequenceGenerator();
Subject<bool> lockReleaser = new Subject<bool>();

Stopwatch watch = new Stopwatch();
watch.Start();
var test = gen.messageQueue.Buffer(TimeSpan.FromSeconds(4),5)
        .Zip(lockReleaser, (res, _) => res)
    .Subscribe(async x =>
    {
        if (x.Count > 0)
        {
            Console.WriteLine($"{watch.Elapsed}: {x[x.Count - 1]}------");
            await Task.Delay(10000);
        }
        lockReleaser .OnNext(true);
        Console.WriteLine($"1:------");
    },
    x => Console.WriteLine("EX"),
    () => Console.WriteLine("1 Done"));
lockReleaser.OnNext(true);

【讨论】：