我正在努力使用.NET响应式技术构建生产者..消费者模式。生产者从Kafka消息总线中读取消息。一旦读取到消息,就需要将其移交给消费者来处理该消息。
我已经能够使用.NET响应式技术完成了这个任务。但是,我注意到消费者在与生产者相同的线程上处理消息。请参见下面的代码。目标是:拥有一个单独的生产者线程,从总线中读取消息。然后,在另一个线程上将其移交给消费者以处理消息。我目前的代码如下:
// Producer Code
private Subject<LGMessage> _onMessageSubject = new Subject<LGMessage>();
private IObserver<LGMessage> messageBusObserver;
protected IObservable<LGMessage> _onMessageObservable
{
get
{
return _onMessageSubject.AsObservable();
}
}
public void AddObserver(IObserver<LGMessage> observer)
{
_onMessageObservable.ObserveOn(NewThreadScheduler.Default).Subscribe(observer);
}
// Read is called when the message is read from the bus
public bool Read(Message<string, string> msg)
{
// add the message to the observable
_onMessageSubject.OnNext(msg.Value);
}
// Consumer Code
public virtual void OnNext(string value)
{
Console.WriteLine("Thread {0} Consuming",
Thread.CurrentThread.ManagedThreadId);
Console.WriteLine("{1}: Message I got from bus: {0}", value.Key,
this.Name);
// Take Action
}