我正在使用StackExchange Redis Client.我的目标是从客户端公开的Pub Sub Subscriber创建一个Observable流,然后可以支持Observables的1-n订阅,每个订阅都有自己的LINQ过滤器. (发布按计划运行,问题纯粹围绕特定频道上的事件流订阅.)
背景：
我使用Redis Pub Sub作为事件源CQRS应用程序的一部分.具体用例是将事件发布给多个订阅者,然后更新各种读取模型,发送电子邮件等.
这些订阅者中的每一个都需要过滤他们处理的事件类型,为此我希望使用Rx .Net(Reactive Extensions)和LINQ,
在事件流上提供过滤条件,以有效地处理仅对感兴趣的事件做出反应.使用这种方法消除了使用事件总线实现注册处理程序的需要,并允许我通过部署1-n微服务向系统添加新投影,每个微服务都有1-n Observables使用自己的特定过滤器订阅事件流.
我做了什么：
1)我创建了一个继承自ObservableBase的类,重写了SubscribeCore方法,该方法接收来自Observables的订阅请求,将它们存储在ConcurrentDictionary中,并且当每个Redis通知从通道到达时,循环通过已注册的Observable订阅者并调用它们的OnNext方法传递RedisValue.
2)我创建了一个Subject,它也接受来自Observables的订阅,并调用它们的OnNext方法.同样,对象的使用似乎被许多人所厌恶.
问题：
我尝试过的方法功能(至少表面上看),具有不同的性能水平,但感觉像是一个黑客,而且我没有按照预期的方式使用Rx.
我看到许多评论认为应尽可能使用内置的Observable方法,例如Observable.FromEvent,但这似乎与StackExchange Redis客户端订阅API无关,至少对我来说是这样.
我也理解接收流并转发到多个观察者的首选方法是使用ConnectableObservable,它似乎是为我面临的场景设计的(每个微服务将在内部订阅1-n Observables).目前,我无法理解如何将ConnectableObservable连接到StackExchange Redis的通知,或者它是否提供了对Observable的真正好处.
更新：
尽管在我的场景中完成不是问题(Disposal很好),但错误处理很重要;例如隔离在一个订户中检测到的错误,以防止所有订阅终止.
解决方法
public static IObservable<RedisValue> WhenMessageReceived(this ISubscriber subscriber,RedisChannel channel)
{
return Observable.Create<RedisValue>(async (obs,ct) =>
{
await subscriber.SubscribeAsync(channel,(_,message) =>
{
obs.OnNext(message);
}).ConfigureAwait(false);
return Disposable.Create(() => subscriber.Unsubscribe(channel));
});
}
由于没有完成Redis通道,因此生成的IObservable永远不会完成,但是您可以删除IDisposable订阅以取消订阅Redis通道(这将由许多Rx操作符自动完成).
用法可能是这样的：
var subscriber = connectionMultiplexer.GetSubscriber();
var gotMessage = await subscriber.WhenMessageReceived("my_channel")
.AnyAsync(msg => msg == "expected_message")
.ToTask()
.ConfigureAwait(false);
或者根据你的例子：
var subscriber = connectionMultiplexer.GetSubscriber();
var sendEmailEvents = subscriber.WhenMessageReceived("my_channel")
.Select(msg => ParseEventFromMessage(msg))
.Where(evt => evt.Type == EventType.SendEmails);
await sendEmailEvents.ForEachAsync(evt =>
{
SendEmails(evt);
}).ConfigureAwait(false);
其他微服务可能会有不同的过滤.
