为了详细说明以前的答案,你可以使用Observable.Create<TResult>的一系列方法来完成想要的功能。 这是一个例子: var observable = Observable.Create<char>(async (observer, cancel) =>
{
for (var i = 0; !cancel.IsCancellationRequested && i < 100; i++)
{
observer.OnNext(await GetCharAsync());
}
});
你也可以在LINQPad中使用它: // Create a disposable that keeps the query running.
// This is necessary, since the observable is 100% async.
var end = Util.KeepRunning();
observable.Subscribe(
c => Console.WriteLine(c.ToString()),
() => end.Dispose());
... 展开详请