5

我正在尝试 Java 9 中的一些新功能。所以我进行了一个测试，以拥有一个发布者，以给定的速率发布数字。我还实现了一个订阅者来收听这些出版物并将它们打印到控制台。

虽然我可能不完全理解如何使用这个 Api，因为该onNext()方法没有打印任何内容，getLastItem()只返回 0。

唯一似乎有效的部分是onSubscribe()正确初始化lastItem变量的部分。

@Test
public void testReactiveStreams(){
    //Create Publisher
    SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

    //Register Subscriber
    TestIntegerSubscriber subscriber = new TestIntegerSubscriber();
    publisher.subscribe(subscriber);

    assertTrue(publisher.hasSubscribers());

    //Publish items
    System.out.println("Publishing Items...");

    List.of(1,2,3,4,5).stream().forEach(i -> {
        publisher.submit(i);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            // blah
        }
    });
    assertEquals(5, subscriber.getLastItem());

    publisher.close();
}


private class TestIntegerSubscriber implements Flow.Subscriber<Integer> {

    private int lastItem;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Subscribed");
        lastItem = 0;
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Received : "+item);
        lastItem += 1; // expect increment by 1
        assertTrue(lastItem == item);
    }

    @Override
    public void onError(Throwable throwable) {
        // nothing for the moment
    }

    @Override
    public void onComplete() {
        System.out.println("Completed");
    }

    public int getLastItem(){
        return lastItem;
    }
}

有人可以告诉我我在测试中做错了什么吗？我希望测试能够打印这些数字并返回 5 作为最后一项。

我不得不说我只在 Angular2 中使用 Observables 和 Subjects，尽管它们似乎更容易理解。

4

2 回答 2

4

Flow API 实现了一个称为背压的特性（这里在 RxJava 的上下文中解释），这意味着发布者不应该通过发布项目的速度超过其处理它们的速度来压倒订阅者。JDK 9 实现这一点的方式是让订阅者从订阅中请求项目。

对于您的测试，TestIntegerSubscriber应该请求项目onSubscription，假设是 10，并跟踪onNext调用频率，以便在推送 10 个项目后可以请求更多。

我写了一个关于 Flow API 的部分，其中更详细一点。它还描述了发布者、订阅者和订阅之间的交互：

  1. 创建一个Publisher和一个Subscriber。
  2. 使用 订阅订阅者Publisher::subscribe。
  3. 发布者创建一个Subscription并调用Subscriber::onSubscription它，以便订阅者可以存储订阅。
  4. 在某些时候，订户呼叫Subscription::request请求一些项目。
  5. 发布者通过调用 开始将项目交给订阅者Subscriber::onNext。它永远不会发布超过请求数量的项目。
  6. 发布者可能在某个时候耗尽或遇到麻烦并分别调用Subscriber::onCompleteor Subscriber::onError。
  7. 订阅者可能会不时地继续请求更多项目，或者通过调用来切断连接Subscription::cancel。
于 2017-03-15T06:38:03.440 回答
0

我做了更多的研究，我可以看到代码可以通过Subscription在实现中添加一个成员来工作，Subscriber如下所示：

public class MySubscriber<T> implements Subscriber<T> {  
  private Subscription subscription;

...

@Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            System.out.println("Subscribed");
            subscription.request(1);
            lastItem = 0;
        }

...

  @Override
        public void onNext(Integer item) {
            System.out.println("Received : "+item);
            lastItem += 1; // expect increment by 1
            assertTrue(lastItem == item);
            subscription.request(1);
        }

subscription.request(1); 我认为这意味着我正在消费出版商的 1 件商品。我不知道……如果有人向我澄清一下，实际上会很好。

于 2017-03-14T19:26:23.133 回答