java - 如何使用 rxjava 替换使用 volatile 变量的线程间通信？

Question

我有一个应用程序，它通过让一个线程在另一个线程检查的某个对象上设置一个 volatile 变量来在线程之间进行大量通信。我发现这很容易出错，我想尝试使用 RxJava 替换它，但在某些情况下我不知道如何转换。

我现在遇到的情况是我有两个线程，我们将一个称为 Controller ，另一个称为测量器。测量器的工作是每 100 毫秒记录一些数量。 Controller 做了很多工作，与应用程序的各个部分进行通信，并且它经常会告诉测量器更改其测量的内容。现在，它通过设置一个 volatile 变量来实现这一点，并且测量器循环的每次迭代都会检查该变量以了解要测量的内容。

测量器不能与 Controller 位于同一线程中，因为测量需要时间，并且 Controller 不能延迟它正在执行的其他工作。

感觉解决方案就像使 Controller 成为一个可观察对象，每当测量器的指令需要更新时，它就会发出一个项目，但我能看到测量器在收到事件时改变其行为的唯一方法是让这些事件的订阅者像以前一样设置 volatile 变量，然后我就一无所获。

我想知道是否可以以某种方式获取 Controller 发出的项目流并将其转换为一个不断重复每个项目的流，直到 Controller 发出不同的项目，然后我可以在测量器每次收到测量值都会进行测量。这是正确的方法吗？如果是，如何将 Controller 发出的项目转换为重复的项目流？

Answer 1

我对 Rx 比较陌生，但我会使用BehaviorSubject。您可以使用distinctUntilChanged()，或将其与计时器Observable结合使用:

    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