Реактивная структура (RX) и работа с событиями Асинхронно

Question

Так что я просто играю с RX и изучаю его. Я начал играть с событиями и хотел знать, как подписаться на события, и обрабатывать результаты в асинхронном режиме. Позвольте мне объяснить, с кодом:

Простой класс, который вызывает события:

public class EventRaisingClass 
{ 
    public event EventHandler<SomeEventArgs> EventOccured; 

    //some other code that raises event... 
} 

public class SomeEventArgs : EventArgs 
{ 
    public SomeEventArgs(int data) 
    { 
     this.SomeArg = data; 
    } 

    public int SomeArg { get; private set; } 
} 

Тогда мой Main:

public static void Main(string[] args) 
{ 
    var eventRaiser = new EventRaisingClass(); 
    IObservable<IEvent<SomeEventArgs>> observable = 
     Observable.FromEvent<SomeEventArgs>(e => eventRaiser.EventOccured += e, e => eventRaiser.EventOccured -= e); 

    IObservable<IList<IEvent<SomeEventArgs>>> bufferedEvents = observable.BufferWithCount(100); 

    //how can I subscribte to bufferedEvents so that the subscription code gets called Async? 
    bufferedEvents.Subscribe(list => /*do something with list of event args*/); //this happens synchrounously... 

} 

Как вы можете видеть в моих комментариях, когда вы просто вызываете подписаться так , весь код подписки происходит синхронно. Есть ли выход из коробки с помощью RX, чтобы вызывать Subscribe в разных потоках всякий раз, когда есть новая партия событий для работы?

Answer 1

Я считаю, вы искали SubscribeOn или ObserveOn, минуя IScheduler. Есть несколько планировщиков, встроенных под System.Concurrency; некоторые из них используют любой поток, текущий, а другие используют конкретные потоки.

This video имеет дополнительную информацию о концепции планировщика.

Команда Rx также недавно выпустила документ hands-on labs, который является самым близким к учебнику прямо сейчас.

Answer 2

bufferedEvents.ObserveOn(Scheduler.TaskPool).Subscribe(... 

SubscribeOn это указать график, на котором так называемые «подписки побочные эффекты» происходят. Например, ваш наблюдаемый может открывать файл каждый раз, когда кто-то подписывается.

ObserveOn - указать расписание, по которому вызов наблюдателя будет происходить каждый раз, когда появляется новое значение. На практике он используется чаще, чем SubscribeOn.