SpringCloud 反應(yīng)源

Spring Cloud Stream反應(yīng)性支持還提供了通過@StreamEmitter注釋創(chuàng)建反應(yīng)性源的功能。通過使用@StreamEmitter批注，可以將常規(guī)源轉(zhuǎn)換為被動(dòng)源。@StreamEmitter是方法級(jí)別的注釋，用于將方法標(biāo)記為用@EnableBinding聲明的輸出的發(fā)射器。您不能將@Input批注與@StreamEmitter一起使用，因?yàn)闃?biāo)有該批注的方法不會(huì)監(jiān)聽任何輸入。而是用標(biāo)記為@StreamEmitter的方法生成輸出。遵循@StreamListener中使用的相同編程模型，@StreamEmitter還允許靈活地使用@Output批注，具體取決于方法是否具有任何參數(shù)，返回類型和其他考慮因素。

本節(jié)的其余部分包含使用各種樣式的@StreamEmitter批注的示例。

以下示例每毫秒發(fā)出一次Hello, World消息，并發(fā)布到Reactor Flux中：

@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {

  @StreamEmitter
  @Output(Source.OUTPUT)
  public Flux<String> emit() {
    return Flux.intervalMillis(1)
            .map(l -> "Hello World");
  }
}

在前面的示例中，Flux中的結(jié)果消息被發(fā)送到Source的輸出通道。

下一個(gè)示例是@StreamEmmitter的另一種形式，它發(fā)送Reactor Flux。以下方法代替返回Flux，而是使用FluxSender從源代碼中以編程方式發(fā)送Flux：

@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {

  @StreamEmitter
  @Output(Source.OUTPUT)
  public void emit(FluxSender output) {
    output.send(Flux.intervalMillis(1)
            .map(l -> "Hello World"));
  }
}

下一個(gè)示例在功能和樣式上與上述代碼段完全相同。但是，它沒有在方法上使用顯式的@Output注釋，而是在方法參數(shù)上使用了注釋。

@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {

  @StreamEmitter
  public void emit(@Output(Source.OUTPUT) FluxSender output) {
    output.send(Flux.intervalMillis(1)
            .map(l -> "Hello World"));
  }
}

本節(jié)的最后一個(gè)示例是使用Reactive Streams Publisher API并利用Spring Integration Java DSL中對(duì)它的支持來編寫反應(yīng)源的另一種方式。以下示例中的Publisher仍在幕后使用Reactor Flux，但是，從應(yīng)用程序角度看，這對(duì)用戶是透明的，并且對(duì)于Spring Integration僅需要響應(yīng)流和Java DSL：

@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {

  @StreamEmitter
  @Output(Source.OUTPUT)
  @Bean
  public Publisher<Message<String>> emit() {
    return IntegrationFlows.from(() ->
                new GenericMessage<>("Hello World"),
        e -> e.poller(p -> p.fixedDelay(1)))
        .toReactivePublisher();
  }
}