Rx.Net GroupBy，如何将多个观察者订阅到特定组
Rx.Net GroupBy, How to subscribe multiple observers to a specific Group
我正在迈入 Rx 世界的第一步，但发现很难获得预期的结果，尤其是使用 GroupBy 运算符时，因此非常感谢任何帮助。
如何将多个观察者订阅到特定组？
我的要求是：
我有一个 DataProvider class，它定期发出 http Api 请求。
http 响应是 List<Item>。每个项目都有一个唯一的 ID 属性.
我需要根据其 Id 将每个项目作为单独的流处理，这看起来像 GroupBy 的情况。
每个组都需要自己的管道，其中：
- 它以特定值开头（StartWith 运算符）
- 它缓冲前一个项目以与当前项目进行比较（Buffer(2,1) 运算符）
- 如果当前 Item 与前一个（Where）不同，则发出当前 Item
结果是 IObservable<Item> 的更改 (ChangeStream)。我不再与特定群体打交道。
我如何才能留在组管道中并允许多个订阅者订阅特定组？
- 观察者可以提前订阅（在 Item.Id 出现在响应流之前和创建组之前）
- 观察者可以延迟订阅（在 Item.Id 出现在响应流中并且群组已创建之后）
- 迟到的订阅者应该收到 Item.Id（重播（1））的最后更改，但我似乎也无法弄清楚这部分内容。
多播特定组的 Rx 方式是什么？任何帮助/建议将不胜感激。我在下面提供了示例代码。
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
namespace RxTest
{
class Program
{
static void Main(string[] args)
{
var dataService = new MockDataService();
// How do I subscribe to a specific group?
// Eg. I am only interested in changes to Items where Id == 1
// Subscribers can be early (before the stream is hot)
var item1Stream = dataService.SubscribeToItem(1);
// There can be multiple subscribers to a group
var item1Stream2 = dataService.SubscribeToItem(1);
Console.WriteLine("Press Any Key to Start");
Console.ReadLine();
dataService.Start();
// Subscribers can be late (Eg. Subscriber to Item Id == 2 after it has emitted items)
Thread.Sleep(2000);
var item2Stream = dataService.SubscribeToItem(2);
// Subscribers can be early (After connect but before the Item Id appears on the Stream (before group creation))
// Eg. Subscribe to group 4 (Group 4 doesn't get created until 20s after connect in this example)
var item4Stream = dataService.SubscribeToItem(4);
// What is the Rx way to Multicast a Group?
Console.WriteLine("Press Any Key to Exit");
Console.ReadLine();
dataService.Stop();
}
}
public class MockDataService
{
private readonly IConnectableObservable<Item> _itemsStream;
private IDisposable _itemsSubscription;
private readonly IObservable<Item> _changeStream;
private IDisposable _changeSubscription;
public MockDataService()
{
// Simulate Http response pipeline.
//// Time: 1s...............10s..............20s.....etc
//// stream: [[1][2]]repeat...[[2][3]]repeat...[[3][4]]repeat...
IObservable<List<Item>> responseStream = Observable.Interval(TimeSpan.FromSeconds(1))
.Take(50)
.Select(tick =>
{
// Every 10 ticks an item drops off the stream and a new one starts
// Every 2 ticks the Item value increases to generate a change.
int rangeStart = Math.Min(((int)tick / 10) + 1, 5);
return Enumerable.Range(rangeStart, 2).Select(id => new Item(id, (int)tick / 2)).ToList();
});
// Flatten the list into IObservable<Item>
//// Time: 1s.............10s............20s.....etc
//// stream: [1][2]repeat...[2][3]repeat...[3][4]repeat...
_itemsStream = responseStream
.SelectMany(list => list)
.Publish();
// Split into groups by Item.Id and process each group for changes
// ChangeStream is an IObservable<Item> that have changes.
_changeStream = _itemsStream
.GroupBy(item => item.Id)
.SelectMany(grp =>
grp
// Pipeline for each group.
.StartWith(new Item(grp.Key, -1)) // Initial item from Db
//.TakeUntil(Item => Item.IsComplete()) // Logic to complete the group
.LogConsoleWithThread($"Group: {grp.Key}")
.Buffer(2, 1)
.Where(buffer => buffer.Count == 2 && buffer[0].HasChanges(buffer[1]))
.Select(buffer => buffer[1])
.LogConsoleWithThread($"Group.Change : {grp.Key}")
// How do I push changes in this group to Zero..Many subscribed Observers?
// I would also like to Replay(1) to all late subscribers to a group.
);
}
/// <summary>
/// How to get the IObservable for a specific group?
/// </summary>
/// <param name="itemId"></param>
/// <returns></returns>
public IObservable<Item> SubscribeToItem(int itemId)
{
// ????
return null;
}
public void Start()
{
_changeSubscription = _changeStream.SubscribeConsole("ChangeStream");
_itemsSubscription = _itemsStream.Connect();
}
public void Stop()
{
_changeSubscription.Dispose();
_itemsSubscription.Dispose();
}
}
public class Item
{
public int Id { get; private set; }
public int Value { get; private set; }
public Item(int id, int value)
{
Id = id;
Value = value;
}
public bool HasChanges(Item compareItem)
{
return this.Value != compareItem.Value;
}
public override string ToString()
{
return $"Item: Id={Id} Value={Value}";
}
}
public static class RxExtensions
{
public static IDisposable SubscribeConsole<T>(this IObservable<T> observable, string name = "")
{
return observable.Subscribe(new ConsoleObserver<T>(name));
}
/// <summary>
/// Logs to the Console the subscriptions and emissions done on/by the observable
/// each log message also includes the thread it happens on
/// </summary>
/// <typeparam name="T">The Observable Type</typeparam>
/// <param name="observable">The Observable to log.</param>
/// <param name="name">An optional name prefix that will be added before each notification</param>
/// <returns></returns>
public static IObservable<T> LogConsoleWithThread<T>(this IObservable<T> observable, string name = "")
{
return Observable.Defer(() =>
{
Console.WriteLine("{0} Subscription happened on Thread: {1}", name, Thread.CurrentThread.ManagedThreadId);
return observable.Do(
x => Console.WriteLine("{0} - OnNext({1}) Thread: {2}", name, x, Thread.CurrentThread.ManagedThreadId),
ex =>
{
Console.WriteLine("{0} - OnError Thread:{1}", name, Thread.CurrentThread.ManagedThreadId);
Console.WriteLine("\t {0}", ex);
},
() => Console.WriteLine("{0} - OnCompleted() Thread {1}", name, Thread.CurrentThread.ManagedThreadId));
});
}
}
/// <summary>
/// An observer that outputs to the console each time the OnNext, OnError or OnComplete occurs
/// </summary>
/// <typeparam name="T"></typeparam>
public class ConsoleObserver<T> : IObserver<T>
{
private readonly string _name;
public ConsoleObserver(string name = "")
{
_name = name;
}
public void OnNext(T value)
{
Console.WriteLine("{0} - OnNext({1})", _name, value);
}
public void OnError(Exception error)
{
Console.WriteLine("{0} - OnError:", _name);
Console.WriteLine("\t {0}", error);
}
public void OnCompleted()
{
Console.WriteLine("{0} - OnCompleted()", _name);
}
}
}
您可能需要专门的发布运营商，因为现有的运营商（Publish、PublishLast 和 Replay）对您的需求来说太窄或太宽泛。因此，您需要使用 Multicast 运算符，该运算符随自定义重播主题一起提供，该主题仅缓冲每个键的最后一个元素。这是此类主题的基本实现：
public class ReplayLastPerKeySubject<T, TKey> : ISubject<T>
{
private readonly Func<T, TKey> _keySelector;
private readonly ReplaySubject<ReplaySubject<T>> _subjects;
private readonly IObservable<T> _mergedSubjects;
private readonly Dictionary<TKey, ReplaySubject<T>> _perKey;
public ReplayLastPerKeySubject(Func<T, TKey> keySelector)
{
_keySelector = keySelector;
_subjects = new ReplaySubject<ReplaySubject<T>>();
_mergedSubjects = _subjects.Merge();
_perKey = new Dictionary<TKey, ReplaySubject<T>>();
}
public void OnNext(T value)
{
var key = _keySelector(value);
ReplaySubject<T> subject;
if (!_perKey.TryGetValue(key, out subject))
{
subject = new ReplaySubject<T>(1);
_perKey.Add(key, subject);
_subjects.OnNext(subject);
}
subject.OnNext(value);
}
public void OnCompleted()
{
// All subjects, inner and outer, must be completed
_subjects.OnCompleted();
_subjects.Subscribe(subject => subject.OnCompleted());
}
public void OnError(Exception error)
{
// Faulting the master (outer) subject is enough
_subjects.OnError(error);
}
public IDisposable Subscribe(IObserver<T> observer)
{
return _mergedSubjects.Subscribe(observer);
}
}
这个实现是基于一个 , written by an RX expert. The original answer uses a Concat observable for subscribing the observers, while this one uses a Merge observable 的答案，所以我不能 100% 确定它的正确性和效率。
有了这样的实施，剩下的就很容易了。您首先创建原始可观察对象的发布版本：
var published = YourObservable
.Multicast(new ReplayLastPerKeySubject<Item, int>(x => x.Id)))
.RefCount();
最后，您可以使用 Where 运算符为特定键创建更改流：
var changeStream13 = published.Where(x => x.Id == 13);
我正在迈入 Rx 世界的第一步，但发现很难获得预期的结果，尤其是使用 GroupBy 运算符时，因此非常感谢任何帮助。
如何将多个观察者订阅到特定组？
我的要求是：
我有一个 DataProvider class，它定期发出 http Api 请求。
http 响应是 List<Item>。每个项目都有一个唯一的 ID 属性.
我需要根据其 Id 将每个项目作为单独的流处理，这看起来像 GroupBy 的情况。
每个组都需要自己的管道，其中：
- 它以特定值开头（StartWith 运算符）
- 它缓冲前一个项目以与当前项目进行比较（Buffer(2,1) 运算符）
- 如果当前 Item 与前一个（Where）不同，则发出当前 Item
结果是 IObservable<Item> 的更改 (ChangeStream)。我不再与特定群体打交道。
我如何才能留在组管道中并允许多个订阅者订阅特定组？
- 观察者可以提前订阅（在 Item.Id 出现在响应流之前和创建组之前）
- 观察者可以延迟订阅（在 Item.Id 出现在响应流中并且群组已创建之后）
- 迟到的订阅者应该收到 Item.Id（重播（1））的最后更改，但我似乎也无法弄清楚这部分内容。
多播特定组的 Rx 方式是什么？任何帮助/建议将不胜感激。我在下面提供了示例代码。
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
namespace RxTest
{
class Program
{
static void Main(string[] args)
{
var dataService = new MockDataService();
// How do I subscribe to a specific group?
// Eg. I am only interested in changes to Items where Id == 1
// Subscribers can be early (before the stream is hot)
var item1Stream = dataService.SubscribeToItem(1);
// There can be multiple subscribers to a group
var item1Stream2 = dataService.SubscribeToItem(1);
Console.WriteLine("Press Any Key to Start");
Console.ReadLine();
dataService.Start();
// Subscribers can be late (Eg. Subscriber to Item Id == 2 after it has emitted items)
Thread.Sleep(2000);
var item2Stream = dataService.SubscribeToItem(2);
// Subscribers can be early (After connect but before the Item Id appears on the Stream (before group creation))
// Eg. Subscribe to group 4 (Group 4 doesn't get created until 20s after connect in this example)
var item4Stream = dataService.SubscribeToItem(4);
// What is the Rx way to Multicast a Group?
Console.WriteLine("Press Any Key to Exit");
Console.ReadLine();
dataService.Stop();
}
}
public class MockDataService
{
private readonly IConnectableObservable<Item> _itemsStream;
private IDisposable _itemsSubscription;
private readonly IObservable<Item> _changeStream;
private IDisposable _changeSubscription;
public MockDataService()
{
// Simulate Http response pipeline.
//// Time: 1s...............10s..............20s.....etc
//// stream: [[1][2]]repeat...[[2][3]]repeat...[[3][4]]repeat...
IObservable<List<Item>> responseStream = Observable.Interval(TimeSpan.FromSeconds(1))
.Take(50)
.Select(tick =>
{
// Every 10 ticks an item drops off the stream and a new one starts
// Every 2 ticks the Item value increases to generate a change.
int rangeStart = Math.Min(((int)tick / 10) + 1, 5);
return Enumerable.Range(rangeStart, 2).Select(id => new Item(id, (int)tick / 2)).ToList();
});
// Flatten the list into IObservable<Item>
//// Time: 1s.............10s............20s.....etc
//// stream: [1][2]repeat...[2][3]repeat...[3][4]repeat...
_itemsStream = responseStream
.SelectMany(list => list)
.Publish();
// Split into groups by Item.Id and process each group for changes
// ChangeStream is an IObservable<Item> that have changes.
_changeStream = _itemsStream
.GroupBy(item => item.Id)
.SelectMany(grp =>
grp
// Pipeline for each group.
.StartWith(new Item(grp.Key, -1)) // Initial item from Db
//.TakeUntil(Item => Item.IsComplete()) // Logic to complete the group
.LogConsoleWithThread($"Group: {grp.Key}")
.Buffer(2, 1)
.Where(buffer => buffer.Count == 2 && buffer[0].HasChanges(buffer[1]))
.Select(buffer => buffer[1])
.LogConsoleWithThread($"Group.Change : {grp.Key}")
// How do I push changes in this group to Zero..Many subscribed Observers?
// I would also like to Replay(1) to all late subscribers to a group.
);
}
/// <summary>
/// How to get the IObservable for a specific group?
/// </summary>
/// <param name="itemId"></param>
/// <returns></returns>
public IObservable<Item> SubscribeToItem(int itemId)
{
// ????
return null;
}
public void Start()
{
_changeSubscription = _changeStream.SubscribeConsole("ChangeStream");
_itemsSubscription = _itemsStream.Connect();
}
public void Stop()
{
_changeSubscription.Dispose();
_itemsSubscription.Dispose();
}
}
public class Item
{
public int Id { get; private set; }
public int Value { get; private set; }
public Item(int id, int value)
{
Id = id;
Value = value;
}
public bool HasChanges(Item compareItem)
{
return this.Value != compareItem.Value;
}
public override string ToString()
{
return $"Item: Id={Id} Value={Value}";
}
}
public static class RxExtensions
{
public static IDisposable SubscribeConsole<T>(this IObservable<T> observable, string name = "")
{
return observable.Subscribe(new ConsoleObserver<T>(name));
}
/// <summary>
/// Logs to the Console the subscriptions and emissions done on/by the observable
/// each log message also includes the thread it happens on
/// </summary>
/// <typeparam name="T">The Observable Type</typeparam>
/// <param name="observable">The Observable to log.</param>
/// <param name="name">An optional name prefix that will be added before each notification</param>
/// <returns></returns>
public static IObservable<T> LogConsoleWithThread<T>(this IObservable<T> observable, string name = "")
{
return Observable.Defer(() =>
{
Console.WriteLine("{0} Subscription happened on Thread: {1}", name, Thread.CurrentThread.ManagedThreadId);
return observable.Do(
x => Console.WriteLine("{0} - OnNext({1}) Thread: {2}", name, x, Thread.CurrentThread.ManagedThreadId),
ex =>
{
Console.WriteLine("{0} - OnError Thread:{1}", name, Thread.CurrentThread.ManagedThreadId);
Console.WriteLine("\t {0}", ex);
},
() => Console.WriteLine("{0} - OnCompleted() Thread {1}", name, Thread.CurrentThread.ManagedThreadId));
});
}
}
/// <summary>
/// An observer that outputs to the console each time the OnNext, OnError or OnComplete occurs
/// </summary>
/// <typeparam name="T"></typeparam>
public class ConsoleObserver<T> : IObserver<T>
{
private readonly string _name;
public ConsoleObserver(string name = "")
{
_name = name;
}
public void OnNext(T value)
{
Console.WriteLine("{0} - OnNext({1})", _name, value);
}
public void OnError(Exception error)
{
Console.WriteLine("{0} - OnError:", _name);
Console.WriteLine("\t {0}", error);
}
public void OnCompleted()
{
Console.WriteLine("{0} - OnCompleted()", _name);
}
}
}
您可能需要专门的发布运营商，因为现有的运营商（Publish、PublishLast 和 Replay）对您的需求来说太窄或太宽泛。因此，您需要使用 Multicast 运算符，该运算符随自定义重播主题一起提供，该主题仅缓冲每个键的最后一个元素。这是此类主题的基本实现：
public class ReplayLastPerKeySubject<T, TKey> : ISubject<T>
{
private readonly Func<T, TKey> _keySelector;
private readonly ReplaySubject<ReplaySubject<T>> _subjects;
private readonly IObservable<T> _mergedSubjects;
private readonly Dictionary<TKey, ReplaySubject<T>> _perKey;
public ReplayLastPerKeySubject(Func<T, TKey> keySelector)
{
_keySelector = keySelector;
_subjects = new ReplaySubject<ReplaySubject<T>>();
_mergedSubjects = _subjects.Merge();
_perKey = new Dictionary<TKey, ReplaySubject<T>>();
}
public void OnNext(T value)
{
var key = _keySelector(value);
ReplaySubject<T> subject;
if (!_perKey.TryGetValue(key, out subject))
{
subject = new ReplaySubject<T>(1);
_perKey.Add(key, subject);
_subjects.OnNext(subject);
}
subject.OnNext(value);
}
public void OnCompleted()
{
// All subjects, inner and outer, must be completed
_subjects.OnCompleted();
_subjects.Subscribe(subject => subject.OnCompleted());
}
public void OnError(Exception error)
{
// Faulting the master (outer) subject is enough
_subjects.OnError(error);
}
public IDisposable Subscribe(IObserver<T> observer)
{
return _mergedSubjects.Subscribe(observer);
}
}
这个实现是基于一个 Concat observable for subscribing the observers, while this one uses a Merge observable 的答案，所以我不能 100% 确定它的正确性和效率。
有了这样的实施，剩下的就很容易了。您首先创建原始可观察对象的发布版本：
var published = YourObservable
.Multicast(new ReplayLastPerKeySubject<Item, int>(x => x.Id)))
.RefCount();
最后，您可以使用 Where 运算符为特定键创建更改流：
var changeStream13 = published.Where(x => x.Id == 13);