Как я могу заменить связь между потоками, используя переменные переменные с rx java?

Question

У меня есть приложение, которое обеспечивает большую связь между потоками, когда один поток устанавливает переменную переменной для некоторого объекта, который проверяет другой поток. Я считаю, что это очень подвержено ошибкам, и я хочу попробовать заменить его, используя Rx Java, но есть некоторые случаи, когда я не могу понять, как конвертировать.

Случай, с которым я борюсь сейчас у меня два потока, давайте назовем один контроллер, а другой - измеритель. Работа замерщика состоит в том, чтобы записывать некоторое количество каждые 100 мс. Контроллер проделывает большую работу, общаясь с различными частями приложения, и время от времени он говорит измерителю изменять то, что он измеряет. Сейчас он делает это, устанавливая переменную, и каждую итерацию l oop измерителя проверяет эту переменную, чтобы увидеть, что измерять.

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

Такое ощущение, что решение - это что-то вроде создания контроллера, наблюдаемого, который будет излучать элемент всякий раз, когда инструкции для измерителя требуют обновления, но Единственный способ, которым я могу видеть, что измеритель может изменить свое поведение при получении события, это заставить подписчика на эти события установить переменную volatile, как и раньше, и тогда я никуда не попал.

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

Answer 1

Я относительно новичок в Rx, но я бы использовал BehaviorSubject. Вы можете использовать differentUntilChanged () или комбинировать его с наблюдаемым таймером:

    public enum Stat { FOO, BAR }

    public class Controller
    {
        private Subject<Stat> statSubject;

        public Controller()
        {
            statSubject = BehaviorSubject.<Stat>create().toSerialized();
        }

        public Observable<Stat> getStatChange()
        {
            return statSubject.distinctUntilChanged();
        }

        public void setStat( Stat stat )
        {
            statSubject.onNext( stat );
        }
    }

    public class Measurer
    {
        public Measurer( Controller controller )
        {
            Observable.timer( 1, TimeUnit.SECONDS, Schedulers.newThread() )
                .repeat()
                .withLatestFrom(
                        controller.getStatChange(),
                        ( __, stat ) -> stat ) // ignore the Long emitted by timer
                .subscribe( this::measureStat );
        }

        private void measureStat( Stat stat )
        {
            switch( stat )
            {
            case FOO:
                measureFoo();
                break;

            default:
                measureBar();
                break;
            }
        }

        private void measureBar()
        {
            System.out.println( "Measuring Bar" );
        }

        private void measureFoo()
        {
            System.out.println( "Measuring Foo" );
        }
    }

    @Test
    public void testMeasureStats() throws InterruptedException
    {
        Controller controller = new Controller();
        controller.setStat( Stat.BAR );

        @SuppressWarnings( "unused" )
        Measurer measurer = new Measurer( controller );

        Thread.sleep( 5000 );

        controller.setStat( Stat.FOO );

        Thread.sleep( 5000 );

        controller.setStat( Stat.BAR );

        Thread.sleep( 5000 );
    }

Вывод:

Measuring Bar
Measuring Bar
Measuring Bar
Measuring Bar
Measuring Foo
Measuring Foo
Measuring Foo
Measuring Foo
Measuring Foo
Measuring Bar
Measuring Bar
Measuring Bar
Measuring Bar
Measuring Bar