EventBus/PubSub vs (реактивные расширения) RX в отношении четкости кода в однопоточном приложении

Ответ 1

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

1. Более декларативные, менее побочные эффекты и менее изменяемое состояние.

Потоки событий способны инкапсулировать логику и состояние, потенциально оставляя ваш код без побочных эффектов и изменяемых переменных.

Рассмотрим приложение, которое подсчитывает нажатия кнопок и отображает количество кликов в качестве метки.

Простое решение JavaFX:

private int counter = 0; // mutable field!!!

Button incBtn = new Button("Increment");
Label label = new Label("0");

incBtn.addEventHandler(ACTION, a -> {
    label.setText(Integer.toString(++counter)); // side-effect!!!
});

Решение ReactFX:

Button incBtn = new Button("Increment");
Label label = new Label("0");

EventStreams.eventsOf(incBtn, ACTION)
        .accumulate(0, (n, a) -> n + 1)
        .map(Object::toString)
        .feedTo(label.textProperty());

Никакая изменчивая переменная не используется, а побочное действие на label.textProperty() скрывается за абстракцией.

В своей магистерской диссертации Eugen Kiss предложил интегрировать ReactFX с Scala. Используя его интеграцию, решение может выглядеть так:

val incBtn = new Button("Increment")
val label = new Label("0")

label.text |= EventStreams.eventsOf(incBtn, ACTION)
    .accumulate(0, (n, a) => n + 1)
    .map(n => n.toString)

Это эквивалентно предыдущему, с дополнительным преимуществом устранения инверсии управления.

2. Средства устранения сбоев и избыточных вычислений. (Только для ReactFX)

Глюки - это временные несоответствия в наблюдаемом состоянии. ReactFX имеет средства для приостановки распространения события до тех пор, пока все обновления для объекта не будут обработаны, избегая как сбоев, так и избыточных обновлений. В частности, посмотрите приостанавливаемые потоки событий, Indicator, InhiBeans и мое сообщение в блоге о InhiBeans. Эти методы основаны на том, что распространение событий синхронно, поэтому не переводите на rxJava.

3. Очистить соединение между производителем событий и потребителем событий.

Шина событий - это глобальный объект, который любой может публиковать и подписываться. Связь между производителем событий и потребителем событий является косвенной и поэтому менее ясной. С реактивными потоками событий связь между производителем и потребителем гораздо более ясна. Для сравнения:

Шина событий:

class A {
    public void f() {
        eventBus.post(evt);
    }
}

// during initialization
eventBus.register(consumer);
A a = new A();

Отношения между a и consumer не ясны, если посмотреть только на код инициализации.

Потоки событий:

class A {
    public EventStream<MyEvent> events() { /* ... */ }
}

// during initialization
A a = new A();
a.events().subscribe(consumer);

Отношения между a и consumer очень явны.

4. События, опубликованные объектом, проявляются в его API.

Используя пример из предыдущего раздела, в примере шины данных a API не сообщает вам, какие события публикуются экземплярами a. С другой стороны, в образце потоков событий a API заявляет, что экземпляры a публикуют события типа MyEvent.

Ответ 2

Я думаю, вам нужно использовать rxjava, потому что он обеспечивает гораздо большую гибкость. Если вам нужен автобус, вы можете использовать перечисление следующим образом:

public enum Events {

  public static PublishSubject <Object> myEvent = PublishSubject.create ();
}

//where you want to publish something
Events.myEvent.onNext(myObject);

//where you want to receive an event
Events.myEvent.subscribe (...);

.

Ответ 3

Согласно моему комментарию выше, JavaFx имеет класс ObservableValue, который соответствует RX Observable (возможно, ConnectableObservable точнее, поскольку он позволяет более одной подписки). Я использую следующий неявный класс для преобразования из RX в JFX, например:

import scala.collection.mutable.Map
import javafx.beans.InvalidationListener
import javafx.beans.value.ChangeListener
import javafx.beans.value.ObservableValue
import rx.lang.scala.Observable
import rx.lang.scala.Subscription

/**
 * Wrapper to allow interoperability bewteen RX observables and JavaFX
 * observables. 
 */
object JfxRxImplicitConversion {
  implicit class JfxRxObservable[T](theObs : Observable[T]) extends ObservableValue[T] { jfxRxObs =>
    val invalListeners : Map[InvalidationListener,Subscription] = Map.empty
    val changeListeners : Map[ChangeListener[_ >: T],Subscription] = Map.empty
    var last : T = _
    theObs.subscribe{last = _}

    override def getValue() : T = last 

    override def addListener(arg0 : InvalidationListener) : Unit = {
      invalListeners += arg0 -> theObs.subscribe { next : T => arg0.invalidated(jfxRxObs) }
    }

    override def removeListener(arg0 : InvalidationListener) : Unit = {
      invalListeners(arg0).unsubscribe
      invalListeners - arg0
    }

    override def addListener(arg0 : ChangeListener[_ >: T]) : Unit = {
      changeListeners += arg0 -> theObs.subscribe { next : T => arg0.changed(jfxRxObs,last,next) }
    }

    override def removeListener(arg0 : ChangeListener[_ >: T]) : Unit = {
      changeListeners(arg0).unsubscribe
      changeListeners - arg0
    }
  }
}

Затем позволяет использовать привязки свойств (например, ScalaFX, но соответствует Property.bind в JavaFX):

new Label {
    text <== rxObs
}

Где rxObs может быть, например:

val rxObs : rx.Observable[String] = Observable.
  interval(1 second).
  map{_.toString}.
  observeOn{rx.lang.scala.schedulers.ExecutorScheduler(JavaFXExecutorService)} 

который является просто счетчиком, который увеличивается каждую секунду. Просто не забудьте импортировать неявный класс. Я не могу себе представить, что это становится чище, чем это!

Вышеупомянутое немного свернуто из-за необходимости использовать планировщик, который хорошо играет с JavaFx. См. этот вопрос для ссылки на суть реализации JavaFXExecutorService. Существует запрос улучшения для scala RX, чтобы сделать это неявным аргументом, поэтому в будущем вам может не понадобиться вызов .observeOn.

Ответ 4

Я изучил одну или две вещи, так как я задал этот вопрос 2 года назад, вот мое текущее понимание (как объяснено в Stephen FRP book):

Оба пытаются помочь описать машину состояний, то есть описать, как изменяется состояние программы в ответ на события.

Ключевое различие между EventBus и FRP композиционностью:

• Что такое композиция?

  • Функциональное программирование является композиционным. Мы все можем согласиться на это. Мы можем взять любую чистую функцию, объединить ее с другими чистыми функциями и получить более сложную чистую функцию.
  • Составность означает, что когда вы объявляете что-то, то на сайте объявления определяется все поведение объявленного объекта.

• FRP - это композиционный способ описания конечного автомата, а event-bus - нет. Зачем?

  • Он описывает конечный автомат.
  • Это композиция, потому что описание выполняется с использованием чистых функций и неизменных значений.

• EventBus не является композиционным способом описания конечного автомата. Почему бы и нет?

  • Вы не можете взять любые два автобуса событий и составить их таким образом, чтобы создать новую сконфигурированную шину событий, описывающую скомпилированный конечный автомат. Почему нет?
    • Банки событий не являются гражданами первого класса (в отличие от события/потока FRP).
      • Что произойдет, если вы попытаетесь сделать автобусы для людей первого класса?
        • Затем вы получаете что-то похожее на FRP/RX.
    • Состояние, на которое влияют автобусы событий, не
      • граждане первого класса (т.е. ссылочно прозрачные, чистые значения в отличие от поведения/ячейки FRP)
      • привязаны к событиям в декларативном/функциональном режиме, вместо этого состояние изменено, принудительно вызванное обработкой событий

Таким образом, EventBus не является композиционным, потому что значение и поведение сконфигурированного EventBus (то есть эволюция времени состояния, на которое влияет упомянутый состав EventBus) зависят от времени (т.е. состояния тех частей программного обеспечения, которые не включены явно в объявлении созданного EventBus). Другими словами, если бы я попытался объявить составной EventBus, тогда было бы невозможно определить (просто взглянув на объявление сконфигурированного EventBus), какие правила определяют эволюцию состояния тех состояний, на которые влияет составленный EventBus, это в отличие от FRP, где это можно сделать.)