Spring Cloud Stream 建立在由企业集成模式定义的概念和模式的基础上，其内部实现依赖于 Spring 项目组合（Spring 集成框架）中已经建立并流行的企业集成模式的实现。

因此，支持 Spring 集成已经建立的基础、语义和配置选项是很自然的。

例如，可以将 Source 的输出通道附加到 MessageSource，并使用熟悉的 @InboundChannelAdapter 注解，如下所示：

@EnableBinding(Source.class)
public class TimerSource {

  @Bean
  @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "10", maxMessagesPerPoll = "1"))
  public MessageSource<String> timerMessageSource() {
    return () -> new GenericMessage<>("Hello Spring Cloud Stream");
  }
}

同样，在为 Processor 绑定约定提供消息处理程序方法的实现时，可以使用 @Transformer 或 @ServiceActivator，如下例所示：

@EnableBinding(Processor.class)
public class TransformProcessor {
  @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
  public Object transform(String message) {
    return message.toUpperCase();
  }
}

	Note
	虽然这可能有点超前，但必须理解的是，当你使用 @StreamListener 注解从同一绑定使用时，会使用 pub-sub 模型。用 @StreamListener 注解的每个方法都会收到自己的消息副本，并且每个方法都有自己的消费者组。但是，如果你使用一个 Spring 集成注解（如 @Aggregator、@Transformer 或 @ServiceActivator）使用同一绑定，那么这些注解将在竞争模型中使用。不会为每个订阅创建单独的使用者组。

作为对 Spring 集成支持的补充，Spring Cloud Stream 提供了自己的 @streamListener 注解，该注解是根据其他 Spring 消息注解（@MessageMapping、@JmsListener、@RabbitListener 和其他注解）建模的，并提供了基于内容的路由和其他方便方法。

@EnableBinding(Sink.class)
public class VoteHandler {

  @Autowired
  VotingService votingService;

  @StreamListener(Sink.INPUT)
  public void handle(Vote vote) {
    votingService.record(vote);
  }
}

与其他 Spring 消息传递方法一样，方法参数可以用 @Payload、@Headers 和 @Header 注解。

对于返回数据的方法，必须使用 @SendTo 注解为该方法返回的数据指定输出绑定目标，如下例所示：

@EnableBinding(Processor.class)
public class TransformProcessor {

  @Autowired
  VotingService votingService;

  @StreamListener(Processor.INPUT)
  @SendTo(Processor.OUTPUT)
  public VoteResult handle(Vote vote) {
    return votingService.record(vote);
  }
}

Spring Cloud Stream 支持将消息分派到多个处理程序方法，这些方法根据条件用 @StreamListener 注解。

为了有资格支持条件分派，方法必须满足以下条件：

条件由注解的 condition 参数中的 SpEL 表达式指定，并对每条消息进行计算。所有与条件匹配的处理程序都在同一个线程中调用，并且不必对调用发生的顺序进行假设。

在具有调度条件的 @StreamListener 的以下示例中，所有具有值 bogey 的头 type 的消息都被调度到 receiveBogey 方法，并且所有具有值 bacall 的头 type 的消息都被调度到 receiveBacall 方法。

@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {

    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'")
    public void receiveBogey(@Payload BogeyPojo bogeyPojo) {
       // handle the message
    }

    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bacall'")
    public void receiveBacall(@Payload BacallPojo bacallPojo) {
       // handle the message
    }
}

condition 上下文中的内容类型协商

使用 @StreamListener 的 condition 参数了解基于内容的路由背后的一些机制是很重要的，尤其是在整个消息类型的上下文中。 如果你在继续之前熟悉小节 32, 内容类型协商，也会有所帮助。

考虑以下情况：

@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class CatsAndDogs {

    @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Dog'")
    public void bark(Dog dog) {
       // handle the message
    }

    @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Cat'")
    public void purr(Cat cat) {
       // handle the message
    }
}

上述代码完全有效。它编译和部署没有任何问题，但是它从来不会产生你期望的结果。

那是因为你正在测试一些还没有在你期望的状态中存在的东西。这是因为消息的有效负载尚未从有线格式（byte[]）转换为所需类型。换句话说，它还没有完成 小节 32, 内容类型协商 中描述的类型转换过程。

因此，除非使用计算原始数据的 SPeL 表达式（例如，字节数组中第一个字节的值），否则请使用基于消息头的表达式（例如，condition = "headers['type']=='dog'"）。

	Note
	目前，通过 @StreamListener 条件进行调度仅支持基于通道的绑定（不支持响应式编程）。

从 Spring Cloud Stream V2.1 开始，定义 stream handlers 和 sources 的另一种选择是使用 Spring Cloud Function 的内置支持，它们可以被表示为 java.util.function.[Supplier/Function/Consumer] 类型的 bean。

要指定要绑定到绑定公开的外部目标的功能 bean，必须提供 spring.cloud.stream.function.definition 属性。

下面是将消息处理程序公开为 java.util.function.Function 的处理器应用程序示例。

@SpringBootApplication
@EnableBinding(Processor.class)
public class MyFunctionBootApp {

	public static void main(String[] args) {
		SpringApplication.run(MyFunctionBootApp.class, "--spring.cloud.stream.function.definition=toUpperCase");
	}

	@Bean
	public Function<String, String> toUpperCase() {
		return s -> s.toUpperCase();
	}
}

在上面的内容中，我们简单地定义了一个名为 toUpperCase 的 java.util.function.Function 类型的 bean，并将其标识为要用作消息处理程序的 bean，其 'input' 和 'output' 必须绑定到由处理器绑定公开的外部目标。

下面是支持源、处理器和接收器的简单功能应用程序的示例。

下面是定义为 java.util.function.Supplier 的源应用程序示例

@SpringBootApplication
@EnableBinding(Source.class)
public static class SourceFromSupplier {
	public static void main(String[] args) {
		SpringApplication.run(SourceFromSupplier.class, "--spring.cloud.stream.function.definition=date");
	}
	@Bean
	public Supplier<Date> date() {
		return () -> new Date(12345L);
	}
}

下面是定义为 java.util.function.Function 的处理器应用程序的示例

@SpringBootApplication
@EnableBinding(Processor.class)
public static class ProcessorFromFunction {
	public static void main(String[] args) {
		SpringApplication.run(ProcessorFromFunction.class, "--spring.cloud.stream.function.definition=toUpperCase");
	}
	@Bean
	public Function<String, String> toUpperCase() {
		return s -> s.toUpperCase();
	}
}

下面是定义为 java.util.function.Consumer 的接收器应用程序的示例

@EnableAutoConfiguration
@EnableBinding(Sink.class)
public static class SinkFromConsumer {
	public static void main(String[] args) {
		SpringApplication.run(SinkFromConsumer.class, "--spring.cloud.stream.function.definition=sink");
	}
	@Bean
	public Consumer<String> sink() {
		return System.out::println;
	}
}

@Bean
public Function<String, String> wrapInQuotes() {
	return s -> "\"" + s + "\"";
}

—spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes

有两种类型的应用程序级错误处理。可以在每个绑定订阅处处理错误，或者全局处理程序可以处理所有绑定订阅错误。让我们回顾一下细节。

图 29.1. 带有自定义和全局错误处理程序的 Spring Cloud Stream 接收器应用程序

对于每个输入绑定，Spring Cloud Stream 使用以下语义创建一个专用的错误通道 <destinationName>.errors。

	Note
	<destinationName> 由绑定的名称（如 input）和组的名称（如 myGroup）组成。

考虑以下事项：

spring.cloud.stream.bindings.input.group=myGroup

@StreamListener(Sink.INPUT) // destination name 'input.myGroup'
public void handle(Person value) {
	throw new RuntimeException("BOOM!");
}

@ServiceActivator(inputChannel = Processor.INPUT + ".myGroup.errors") //channel name 'input.myGroup.errors'
public void error(Message<?> message) {
	System.out.println("Handling ERROR: " + message);
}

在前面的示例中，目标名称是 input.myGroup，而专用的错误通道名称是 input.myGroup.errors。

	Note
	@StreamListener 注解的使用专门用于定义连接内部通道和外部目标的绑定。考虑到特定于目的地的错误通道没有关联的外部目的地，这种通道是 Spring 集成（SI）的特权。这意味着，必须使用一个 SI 处理程序注解（即 @ServiceActivator、@Transformer 等）定义此类目标的处理程序。

	Note
	如果未指定 group，则使用匿名组（如 input.anonymous.2K37rb06Q6m2r51-SPIDDQ），这不适用于错误处理场景，因为在创建目标之前你不知道它将是什么。

此外，如果你绑定到现有目标，例如：

spring.cloud.stream.bindings.input.destination=myFooDestination
spring.cloud.stream.bindings.input.group=myGroup

完整的目标名称是 myFooDestination.myGroup，然后专用的错误通道名称是 myFooDestination.myGroup.errors。

回到示例…​

订阅名为 input 的通道的 handle(..) 方法引发异常。如果还存在错误通道 input.myGroup.errors 的订阅服务器，则所有错误消息都由该订阅服务器处理。

如果有多个绑定，则可能需要一个错误处理程序。Spring Cloud Stream 通过将每个错误通道桥接到名为 errorChannel 的通道自动提供对全局错误通道的支持，允许单个订阅处理所有错误，如下例所示：

@StreamListener("errorChannel")
public void error(Message<?> message) {
	System.out.println("Handling ERROR: " + message);
}

如果错误处理逻辑相同，不管哪个处理程序产生了错误，这可能是一个方便的选项。

系统级的错误处理意味着错误被传回消息传递系统，并且，考虑到并非每个消息传递系统都是相同的，不同的绑定器之间的功能可能不同。

也就是说，在本节中，我们将解释系统级错误处理背后的一般思想，并以 Rabbit 绑定器为例。注意：Kafka 绑定器提供了类似的支持，尽管一些配置属性确实不同。此外，有关详细信息和配置选项，请参阅各个绑定器的文档。

如果没有配置内部错误处理程序，则错误将传播到绑定器，绑定器随后将这些错误传播回消息传递系统。根据消息传递系统的功能，此类系统可能会丢弃消息、重新排队以重新处理消息或将失败消息发送给 DLQ。Rabbit 和 Kafka 都支持这些概念。但是，其他绑定器可能不会，因此有关支持的系统级错误处理选项的详细信息，请参阅单个绑定器的文档。

spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true

delivery_mode:	1
headers:
x-death:
count:	1
reason:	rejected
queue:	input.hello
time:	1522328151
exchange:
routing-keys:	input.myGroup
Payload {"name”:"Bob"}

spring.cloud.stream.rabbit.bindings.input.consumer.republish-to-dlq=true

delivery_mode:	2
headers:
x-original-exchange:
x-exception-message:	has an error
x-original-routingKey:	input.myGroup
x-exception-stacktrace:	org.springframework.messaging.MessageHandlingException: nested exception is
      org.springframework.messaging.MessagingException: has an error, failedMessage=GenericMessage [payload=byte[15],
      headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=input.hello, amqp_deliveryTag=1,
      deliveryAttempt=3, amqp_consumerQueue=input.hello, amqp_redelivered=false, id=a15231e6-3f80-677b-5ad7-d4b1e61e486e,
      amqp_consumerTag=amq.ctag-skBFapilvtZhDsn0k3ZmQg, contentType=application/json, timestamp=1522327846136}]
      at org.spring...integ...han...MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:107)
      at. . . . .
Payload {"name”:"Bob"}

spring.cloud.stream.bindings.input.consumer.max-attempts=1
spring.cloud.stream.rabbit.bindings.input.consumer.requeue-rejected=true

RetryTemplate 是 Spring Retry 库的一部分。虽然覆盖 RetryTemplate 的所有功能超出了本文档的范围，但我们将提到与 RetryTemplate 特定相关的以下消费者属性：

虽然前面的设置足以满足大多数自定义需求，但它们可能无法满足某些复杂的需求，此时你可能需要提供自己的 RetryTemplate 实例。为此，在应用程序配置中将其配置为 bean。应用程序提供的实例将覆盖框架提供的实例。此外，为了避免冲突，必须将绑定器要使用的 RetryTemplate 实例限定为 @StreamRetryTemplate。例如，

@StreamRetryTemplate
public RetryTemplate myRetryTemplate() {
    return new RetryTemplate();
}

从上面的示例中可以看到，由于 @StreamRetryTemplate 是一个合格的 @Bean，所以不需要用 @Bean 对其进行注解。

基于 Reactor 的处理程序可以具有以下参数类型：

基于 Reactor 的处理程序支持 Flux 的返回类型。在这种情况下，它必须用 @Output 注解。我们建议在单输出 Flux 可用时使用该方法的返回值。

以下示例显示了基于 Reactor 的 Processor：

@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {

  @StreamListener
  @Output(Processor.OUTPUT)
  public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
    return input.map(s -> s.toUpperCase());
  }
}

使用输出参数的同一个处理程序如下所示：

@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {

  @StreamListener
  public void receive(@Input(Processor.INPUT) Flux<String> input,
     @Output(Processor.OUTPUT) FluxSender output) {
     output.send(input.map(s -> s.toUpperCase()));
  }
}

Spring Cloud Stream 响应式主持还提供了通过 @StreamEmitter 注解创建响应式源的能力。通过使用 @StreamEmitter 注解，可以将常规源转换为响应式源。@StreamEmitter 是一个方法级注解，它将方法标记为使用 @EnableBinding 声明的输出的发射器。不能将 @Input 注解与 @StreamEmitter 一起使用，因为用此注解标记的方法没有监听任何输入。相反，标记为 @StreamEmitter 的方法会生成输出。按照 @StreamListener 中使用的相同编程模型，@StreamEmitter 还允许灵活地使用 @Output 注解，这取决于该方法是否具有任何参数、返回类型和其他注意事项。

本节中的剩余部分包括使用 @StreamEmitter 注解的不同风格。

以下示例每毫秒发出 Hello, World 消息，并发布到一个 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 中产生的消息被发送到 Source 的输出通道。

下一个例子是发送 Reactor Flux 的 @StreamEmmitter 的另一种风格。以下方法使用 FluxSender 以编程方式从源发送 Flux，而不是返回 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"));
  }
}

下一个例子在功能和风格上与上面的片段完全相同。但是，它不在方法上使用显式的 @Output 注解，而是在方法参数上使用注解。

@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"));
  }
}

本节中的最后一个示例是使用 Reactive Streams Publisher API 和利用 Spring Integration Java DSL 对它的支持的另一种编写响应式源的特色。下面的示例中的 Publisher 仍然使用上面的 Reactor Flux，但是，从应用的角度来看，这对于用户是透明的，并且只需要 Reactive Stream 和 Java DSL 用于 Spring集成：

@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();
  }
}