[Spring Cloud]Stream组件介绍-天天报资讯

2023-03-15 01:09:03 来源：腾讯云

SCS 在 3.x 做了很大的改动，废除了诸如 @StreamListener、@Input、@Output 等类，保留了 Binder、Binding，并提供了批量消费的支持。本着学新不学旧的原则，本文将介绍 SCS 3.x 相关内容。由于关于 spring cloud stream kafka 的文档比较充足，本文就此为例介绍 SCS。

【资料图】

Binder 是提供与外部消息中间件集成的组件，为 Binding 提供了 2 个方法，分别是 bindConsumer 和 bindProducer，它们用于构造生产者和消费者。 Binding 是连接应用程序跟消息中间件的桥梁，用于消息的消费和生产。

Binder 事务

不要在事务中尝试重试和提交死信。重试时，事务可能已经回归。如果想要提交死信用于善后，那么可以使用 DefaultAfterRollbackProcessor以在回滚之后提交死信。

Error Channel

binder 会使用 Error Channel 向消费者传递异常，同时可以配置异步生产者发生异常时将异常传递到 Error Channel。

Dead-Letter

默认情况下，某 topic 的死信队列将与原始记录存在于相同分区中。

死信队列中的消息是允许复活的，但是应该避免消息反复消费失败导致多次循环进入死信队列。

应该使用一个专门的处理程序用来对这些死信队列的信息进行善后。

Consumer 消费者

顾名思义，Consumer 定义的是一个消费者，他是一个函数式接口，提供了消费消息的方法。我们可以直接在 Bean 声明中使用 lambda 表达式实现它。

值得注意的是，Consumer 还是一个泛型接口，通过泛型来绑定消息的类型。接收消息的类型我们会用到 KStream 类，他将与发送消息时定义的 KStream 对应，是键值对组成的抽象记录流，但相同 key 的记录不会被覆盖。

@Beanpublic Consumer> consumer() {return input -> input.foreach((key, value) -> {do consume;});}

当我们在应用程序中声明返回 Consumer 的 Bean，那么这个 Bean 就会自动接入消息队列。另外，我们需要用到 spring.cloud.stream.bindings.{beanName}-in-{idx}={topic}来设置订阅的消息主题。默认情况下，topic 与 beanName 同名。

spring.cloud.stream.bindings.consumer-in-0 = userBuy

当接收到消息时，就会调用 Consumer 定义的 accept 方法进行消息消费。

发送消息生产者

SCS 并没有对发送消息做出一个具体的封装，而是建议通过各个消息队列支持的 client 或者 template 发送消息。

kafkaTemplate.send(message);

Function 加工厂

但有时候，我们需要对数据进行加工后发送回消息队列中，这个时候就会用到 Function。

它和 Consumer 类似，但是方法多了一个返回值。同样的，这个返回值需要用到 KStream 类，这样就能够支持将处理完的数据返回到消息队列。

@Beanpublic Function, KStream> processor() {return input -> input.map((key, value) -> {do process;return new KeyValue(key, value);})}

spring.cloud.stream.bindings.{beanName}-out-{idx}={topic}来设置出口的消息主题。默认情况下，topic 与 beanName 同名。

Function 相比生产者或消费者，更像是将消息进行加工，这个过程可以对消息进行一系列的处理，包括消息拆分，消息过滤和计算中间结果等。常见的一个用途就是国际化消息和多平台通知。

国际化消息就是对消息进行本地化，Function 就类似一个翻译官的功能，将翻译好的消息转达给消费者。
有时候我们也需要同时对多个平台推送通知，比如邮件、短信等。一般来说，邮件服务器和短信服务器不会写死消息的模板以提高泛用性，这个时候就需要中间人对消息进行加工，嵌入对应平台的模板。

多输出绑定

上面提到了消息拆分，Function 允许多个 topic 的消息发送，返回值上会用到 KStream 数组，然后配置上会用到方才展示的 spring.cloud.stream.bindings.{beanName}-out-{idx}={topic}，idx 代表的就是返回值 KStream 在数组中的索引。