system.reactive 相关问题

System.Reactive是指.NET的Reactive Extensions，也称为Rx。 Rx为开发人员提供了通用IObservable <T>接口的反应式编程模型，而不是传统的命令式编程模型或严格依赖.NET事件或特定API的其他反应式编程模型。

如何为一组可观察对象实现 Observable.Publish

我正在寻找实现以下功能：公共静态 IObservable 发布( 这个 IObservable[] 来源，函数我正在寻找实现以下功能： public static IObservable<TResult> Publish<TSource, TResult>( this IObservable<TSource>[] sources, Func<IObservable<TSource>[], IObservable<TResult>> selector) { // how? } 行为应该与现有的相同 IObservable<TResult> Publish<TSource, TResult>( IObservable<TSource> source, Func<IObservable<TSource>, Iobservable<TResult>> selector) 当然唯一的区别是源（和选择器的参数）是一个可观察数组。基本上，“给定一组可观察对象，产生一个结果可观察对象，同时共享对数组的每个可观察对象的单个订阅”。这可以用任何巧妙的方式完成吗？到目前为止，我唯一的想法是通过链接单个可观察的 .Publish() 调用的表达式树来实现这一点——出于性能原因，这是不可行的。或者去做这样的事情。但这看起来很乱，我发现很难推理出正确性 public static IObservable<TResult> Publish<TSource, TResult>( this IObservable<TSource>[] sources, Func<IObservable<TSource>[], IObservable<TResult>> selector) { return Observable.Create<TResult>(observer => { var published = sources.Select(x => x.Publish()).ToArray(); var result = new CompositeDisposable(); result.Add(selector(published).Subscribe(observer)); foreach(var p in published) result.Add(p.Connect()); return result; }); } 我现在想出了一个解决方案，但很高兴听到更好的想法！ public static IObservable<TResult> Publish<TSource, TResult>(this IReadOnlyList<IObservable<TSource>> sources, Func<IReadOnlyList<IObservable<TSource>>, IObservable<TResult>> selector) { IObservable<TResult> Generator(ImmutableArray<IObservable<TSource>> published) { return published.Length == sources.Count ? selector(published) : sources[published.Length].Publish(x => Generator(published.Add(x))); } return Generator(ImmutableArray<IObservable<TSource>>.Empty); } 为什么这个有用的问题出现了。为了清楚起见，我附上了我打算用它做的事情之一。以下函数与现有的 CombineLatest() 工作方式相同，但集成了节流功能：项目的第一个组合会立即发出，之后的所有内容都会去抖动，如果结果选择器的调用开销很大，这会很有帮助。 public static IObservable<TResult> CombineLatest<TSource, TResult>(this IReadOnlyList<IObservable<TSource>> sources, Func<IList<TSource>, TResult> resultSelector, TimeSpan timeSpan, IScheduler? scheduler = null) { scheduler ??= Scheduler.Default; return sources.Publish<TSource, TResult>(publishedSources => publishedSources .CombineLatest(_ => Unit.Default) .Publish(x => x.Take(1).Merge(x.Skip(1).Sample(timeSpan, scheduler)) .Publish(signal => publishedSources .Select(source => signal.WithLatestFrom(source, (_,value) => value)) .Zip() .Select(resultSelector))); } 在我看来，您使用的selector是错误的。这是您当前的实现selector： public static IObservable<TResult> Publish<TSource, TResult>( this IObservable<TSource>[] sources, Func<IObservable<TSource>[], IObservable<TResult>> selector) => selector(sources); 这是微不足道的。这也有点同义反复。你的 selector 是你想要创造的东西。唯一明智的选择器是Func<IObservable<TSource>, IObservable<TResult>> selector. 可以这样吗： public static IObservable<TResult> Publish<TSource, TResult>( this IObservable<TSource>[] sources, Func<IObservable<TSource>, IObservable<TResult>> selector) => sources.Select(x => x.Publish(selector)).Merge();

c# system.reactive

回答 2 投票 0

使用 AtomicIntegers 构造的可观察对象在 RxJava 中的意外行为

这是测试用例：导入 io.reactivex.rxjava3.core.*; 导入 java.util.concurrent.TimeUnit；导入 java.util.concurrent.atomic.AtomicInteger；公共课 MainTest { 公共静态 AtomicInte ...

rx-java atomic system.reactive

回答 1 投票 0

System.Reactive：实现一个 IObservable<T>

我需要创建一个自定义的 IObservable。我在那里读了一些书，最后我不应该直接实施 IObservable。我注意到有一个 ObservableBase。这是……

c# system.reactive

回答 2 投票 0

使用计时器和 Rx 生成整数列表

这对其他人来说似乎是一个基本问题，但由于我刚刚开始使用 Rx，我想知道是否有人可以指出正确的方向。我正在尝试订阅

.net c#-4.0 system.reactive

回答 2 投票 0

将异步函数传递给类型为 Action<X> 的参数？ [重复]

应用有如下定义 IDisposable IObservable.Subscribe(Action onNext) 它被传递了一个异步函数：异步任务流程（X通知，