Reactive Streams

Reactive Streams1.概述在本文中,我们将介绍Java9ReactiveStreams。简而言之,我们将能够使用Flow类,它包含用于构建反应流处理逻辑的主要构建块。ReactiveStreams是具有无阻塞背压的异步流处理的标准。此规范在ReactiveManifesto中定义,并且有各种实现,例如,RxJava或Akka-Streams。2.ReactiveAPI概述要构建Flow,我们可…

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全家桶1年46,售后保障稳定

1.概述
在本文中,我们将介绍Java 9 Reactive Streams。简而言之,我们将能够使用Flow类,它包含用于构建反应流处理逻辑的主要构建块。
Reactive Streams是具有无阻塞背压的异步流处理的标准。此规范在Reactive Manifesto中定义,并且有各种实现,例如,RxJava或Akka-Streams。

2.Reactive API概述
要构建Flow, 我们可以使用三个主要抽象并将它们组合成异步处理逻辑。
每个Flow都需要处理Publisher实例发布给它的事件 ; he Publisher 有一个方法 – subscribe().
如果任何订阅者想要接收由其发布的事件,则他们需要订阅给定的发布者。
消息接收者需要实现订阅者接口。通常,这是每个Flow处理的结束,因为它的实例不会进一步发送消息。我们可以将Subscriber视为一个接收器。这有四个需要重写的方法 – onSubscribe(),onNext(),onError()和onComplete()。
如果我们想要转换传入消息并将其进一步传递给下一个订阅服务器,我们需要实现处理器接口。
这既可以作为订阅者,也可以作为发布者,因为它处理这些消息并将其发送以进行进一步处理。

3.发布和使用消息
假设我们要创建一个简单的Flow,其中我们有一个Publisher发布消息,一个简单的订阅者在消息到达时消费一个消息。让我们创建一个EndSubscriber类。我们需要实现Subscriber接口。接下来,我们将覆盖所需的方法。所述onSubscribe()方法在处理开始之前调用。订阅的实例作为参数传递。它是一个用于控制订阅服务器和发布服务器之间的消息流的类:

public class EndSubscriber<T> implements Subscriber<T> {
    private Subscription subscription;
    public List<T> consumedElements = new LinkedList<>();
 
    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }
}

Jetbrains全家桶1年46,售后保障稳定

我们还初始化一个空列表的consumedElements那将在测试中使用。
现在,我们需要从Subscriber接口实现其余方法。这里的主要方法是onNext() – 每当Publisher发布一条新消息时都会调用它:

  @Override
public void onNext(T item) {
    System.out.println("Got : " + item);
    subscription.request(1);
}

请注意,当我们在onSubscribe()方法中启动订阅时,当我们处理消息时,我们需要在Subscription上调用request()方法,以表示当前订阅服务器已准备好使用更多消息。
最后,我们需要实现onError() – 只要在处理中抛出一些异常就会调用它,以及onComplete() -当Publisher关闭时调用:

@Override
public void onError(Throwable t) {
    t.printStackTrace();
}
 
@Override
public void onComplete() {
    System.out.println("Done");
}

让我们为Processing Flow编写一个测试。我们将使用SubmissionPublisher类 – 来自java.util.concurrent的构造- 它实现了Publisher接口。

@Test
public void whenSubscribeToIt_thenShouldConsumeAll() 
  throws InterruptedException {
  
    // given
    SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    EndSubscriber<String> subscriber = new EndSubscriber<>();
    publisher.subscribe(subscriber);
    List<String> items = List.of("1", "x", "2", "x", "3", "x");
 
    // when
    assertThat(publisher.getNumberOfSubscribers()).isEqualTo(1);
    items.forEach(publisher::submit);
    publisher.close();
 
    // then
     await().atMost(1000, TimeUnit.MILLISECONDS)
       .until(
         () -> assertThat(subscriber.consumedElements)
         .containsExactlyElementsOf(items)
     );
}

注意,我们在EndSubscriber的实例上调用close()方法。它将在给定Publisher的每个订阅服务器下面调用onComplete()回调。

运行该程序将产生以下输出:

Got : 1
Got : x
Got : 2
Got : x
Got : 3
Got : x
Done

4.消息的转换
假设我们想要在发布服务器和订阅服务器之间构建类似的逻辑。我们将创建实现Processor并扩展SubmissionPublisher的TransformProcessor类。
我们将传入一个将输入转换为输出的函数:

public class TransformProcessor<T, R> 
  extends SubmissionPublisher<R> 
  implements Flow.Processor<T, R> {
 
    private Function<T, R> function;
    private Flow.Subscription subscription;
 
    public TransformProcessor(Function<T, R> function) {
        super();
        this.function = function;
    }
 
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }
 
    @Override
    public void onNext(T item) {
        submit(function.apply(item));
        subscription.request(1);
    }
 
    @Override
    public void onError(Throwable t) {
        t.printStackTrace();
    }
 
    @Override
    public void onComplete() {
        close();
    }
}

现在让我们编写一个快速测试,其中包含Publisher发布String元素的处理流程。
我们的TransformProcessor将String解析为Integer – 这意味着需要在此处进行转换:

@Test
public void whenSubscribeAndTransformElements_thenShouldConsumeAll()
  throws InterruptedException {
  
    // given
    SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    TransformProcessor<String, Integer> transformProcessor 
      = new TransformProcessor<>(Integer::parseInt);
    EndSubscriber<Integer> subscriber = new EndSubscriber<>();
    List<String> items = List.of("1", "2", "3");
    List<Integer> expectedResult = List.of(1, 2, 3);
 
    // when
    publisher.subscribe(transformProcessor);
    transformProcessor.subscribe(subscriber);
    items.forEach(publisher::submit);
    publisher.close();
 
    // then
     await().atMost(1000, TimeUnit.MILLISECONDS)
       .until(() -> 
         assertThat(subscriber.consumedElements)
         .containsExactlyElementsOf(expectedResult)
     );
}

注意,在基本Publisher上调用close()方法将导致调用TransformProcessor上的onComplete()方法。

请注意,处理链中的所有Publisher都需要以这种方式关闭。

5.使用订阅控制消息需求
让我们修改我们的EndSubscriber只消耗N个消息。我们将传递该数字作为howMuchMessagesConsume构造函数参数:

public class EndSubscriber<T> implements Subscriber<T> {
  
    private AtomicInteger howMuchMessagesConsume;
    private Subscription subscription;
    public List<T> consumedElements = new LinkedList<>();
 
    public EndSubscriber(Integer howMuchMessagesConsume) {
        this.howMuchMessagesConsume 
          = new AtomicInteger(howMuchMessagesConsume);
    }
 
    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }
 
    @Override
    public void onNext(T item) {
        howMuchMessagesConsume.decrementAndGet();
        System.out.println("Got : " + item);
        consumedElements.add(item);
        if (howMuchMessagesConsume.get() > 0) {
            subscription.request(1);
        }
    }
    //...
     
}

让我们编写一个测试,我们只想从给定的Subscription中使用一个元素:

@Test
public void whenRequestForOnlyOneElement_thenShouldConsumeOne()
  throws InterruptedException {
  
    // given
    SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    EndSubscriber<String> subscriber = new EndSubscriber<>(1);
    publisher.subscribe(subscriber);
    List<String> items = List.of("1", "x", "2", "x", "3", "x");
    List<String> expected = List.of("1");
 
    // when
    assertThat(publisher.getNumberOfSubscribers()).isEqualTo(1);
    items.forEach(publisher::submit);
    publisher.close();
 
    // then
    await().atMost(1000, TimeUnit.MILLISECONDS)
      .until(() -> 
        assertThat(subscriber.consumedElements)
       .containsExactlyElementsOf(expected)
    );
}

虽然发布者发布了六个元素,但我们的EndSubscriber只消耗一个元素,因为它表示只处理那个元素的需求。

六,结论
我们了解了如何创建由发布服务器和订阅服务器组成的处理流程。我们使用处理器转换元素创建了一个更复杂的处理流程。最后,我们使用Subscription来控制订阅者对元素的需求。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请联系我们举报,一经查实,本站将立刻删除。

发布者:全栈程序员-站长,转载请注明出处:https://javaforall.net/222988.html原文链接:https://javaforall.net

(0)
全栈程序员-站长的头像全栈程序员-站长


相关推荐

  • 排序二叉树的实现

    排序二叉树的实现在计算机科学中,二叉树是一种重要的非线性的数据结构。每个结点的度均小于等于2,通常子树称为左子树和右子树。而排序二叉树是二叉树中的一种,其满足:1.如左子树不为空,那么左子树上的结点的值都小于其根上的值;2.如右子树不为空,那么右子树上的结点的值都大于其根上的值;3.其子树也是一个排序二叉树。下面用递归的方式来插入一个结点来满足上述的要求:typedefstructNode{

    2022年7月25日
    10
  • Java 贪吃蛇 最简易地实现方法 超简短代码实现 (附源码 详解)

    Java 贪吃蛇 最简易地实现方法 超简短代码实现 (附源码 详解)分享一个我刚刚写的贪吃蛇小游戏,非常简短的代码实现方法。这可能是全网最简单实现方法,效果图如下:有多简短呢?上代码!!一共四个类!Food类packagecom.snake;importjava.awt.Color;importjava.awt.Graphics;importjava.awt.Point;publicclassFoo…

    2022年6月18日
    35
  • 二级导航菜单[通俗易懂]

    二级导航菜单[通俗易懂]本文用html5+css实现了二级导航菜单,二级导航菜单在网站建设中使用的越来越广泛。效果图如下:当鼠标悬停在一级菜单上时,出现二级下拉菜单二级下拉菜单可以被选中,当鼠标悬停上去时,变色。html代码&lt;!DOCTYPEhtml&gt;&lt;html&gt;&lt;head&gt;&lt;metacharset="UTF-…

    2022年7月26日
    10
  • 拉姆达表达式分组_lomd表达式

    拉姆达表达式分组_lomd表达式List<Map<String,Object>>lpsgjdList=syDkKfslyxxManager.getAlpsgjdQryData(syDkKfslyxxVo); Map<Object,List<Map<String,Object>>>dongh=lpsgjdList.stream().collect( Collectors.groupingBy(e->e.get(“dongh”))); f

    2022年9月19日
    2
  • realsense深度图像保存方法

    realsense深度图像保存方法一般使用realsense时会保存视频序列,当保存深度图像时,需要注意保存的图像矩阵的格式,不然可能造成深度值的丢失。在众多图像库中,一般会使用opencv中的imwrite()函数进行深度图像的保存。一般深度图像中深度值的单位是mm,因此一般使用np.uint16作为最终数据格式保存。例子:importnumpyasnpimportcv2deffun1(…

    2022年4月25日
    31
  • 网络传真机——让传真,超越纸的限制[通俗易懂]

    网络传真机——让传真,超越纸的限制[通俗易懂]网络传真机,离你不再遥远!网络传真-真实号码版,超低价来袭,江苏电信倾力投资,立足江苏,面向全国!13万靓号任君选择。 (1)传真发送功能 点击写传真可进行单个号码发送传真也可多个号码…

    2022年6月28日
    30

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

关注全栈程序员社区公众号