自Spring Cloud Stream的3.1版本以来，EnableBinding、输出和输入已被弃用

Question

自3.1版以来，用于处理队列的主要API已被弃用。课堂评论中写道：

从3.1开始就不推荐使用函数式编程模型

我在网上搜索了很多解决方案，但没有找到关于如何迁移的可靠的E2E解释。

正在寻找以下方面的示例：

1. 从队列读取

如果有几种方法可以做到这一点（正如我在网上看到的），我很乐意为每个选项提供解释和典型用例。

Answer 1

1. 我假设您已经熟悉了主要概念，并将重点关注迁移

首先，一些参考资料可能会有所帮助：

• 这是最初的相关文档：link
• 这是对新功能格式命名方案的解释：link
• 这是一些更高级场景的更详细解释：链接

TL； DR

Spring现在不再使用基于注释的配置，而是使用检测到的消费者/功能/供应商的bean来为您定义流。

输入/消费

而之前的代码是这样的：

interface BindableGradesChannel {
    @Input
    fun gradesChannel(): SubscribableChannel

    companion object {
        const val INPUT = "gradesChannel"
    }
}

其用法类似于：

@Service
@EnableBinding(BindableGradesChannel::class)
class GradesListener {
    private val log = LoggerFactory.getLogger(GradesListener::class.java)
    
    @StreamListener(BindableScoresChannel.INPUT)
    fun listen(grade: Grade) {
        log.info("Received $grade")
        // do something
    }
}

现在，整个定义都无关紧要，可以这样做：

@Service
class GradesListener {
    private val log = LoggerFactory.getLogger(GradesListener::class.java)

    @Bean
    fun gradesChannel(): Consumer<Grade> {
        return Consumer { listen(grade = it) }
    }
    
    fun listen(grade: Grade) {
        log.info("Received $grade")
        // do something
    }
}

请注意，消费者bean是如何替换@StreamListener和@Input的。

关于配置，如果之前为了配置，您有一个应用程序。yml看起来是这样的：

spring:
  cloud:
    stream:
      bindings:
        gradesChannel:
          destination: GradesExchange
          group: grades-updates
          consumer:
            concurrency: 10
            max-attempts: 3

现在应该是这样：

spring:
  cloud:
    stream:
      bindings:
        gradesChannel-in-0:
          destination: GradesExchange
          group: grades-updates
          consumer:
            concurrency: 10
            max-attempts: 3

请注意gradesChannel是如何被gradesChannel-in-0取代的——要了解完整的命名约定，请参阅顶部的命名约定链接。

一些细节：

1. 如果您的应用程序中有多个这样的bean，您需要定义spring.cloud.function.definition属性。
2. 您可以选择给您的通道自定义名称，所以如果您想继续使用gradesChannel，您可以设置spring.cloud.stream.function.bindings.gradesChannel-in-0=gradesChannel，并在配置gradesChannel.

输出/供应商

这里的概念类似，您可以替换如下所示的配置和代码：

interface BindableStudentsChannel {
    @Output
    fun studentsChannel(): MessageChannel
}

和

@Service
@EnableBinding(BindableStudentsChannel::class)
class StudentsQueueWriter(private val studentsChannel: BindableStudentsChannel) {
    fun publish(message: Message<Student>) {
        studentsChannel.studentsChannel().send(message)
    }
}

现在可以替换为：

@Service
class StudentsQueueWriter {
    @Bean
    fun studentsChannel(): Supplier<Student> {
        return Supplier { Student("Adam") }
    }
}

正如你所看到的，我们有一个主要的区别——它是什么时候被调用的，由谁调用的？

以前我们可以手动触发它，但现在它是由spring每秒（默认情况下）触发的。对于需要每秒发布传感器数据的用例来说，这很好，但对于想要发送事件消息的用例来说，这并不好。除了出于任何原因使用函数，spring还提供了两种选择：

StreamBridge-link

使用StreamBridge，您可以像这样显式地定义目标：

@Service
class StudentsQueueWriter(private val streamBridge: StreamBridge) {
    fun publish(message: Message<Student>) {
        streamBridge.send("studentsChannel-out-0", message)
    }
}

通过这种方式，您不需要将目标通道定义为bean，但仍然可以发送消息。缺点是类中有一些显式配置。

ReactorAPI-link

另一种方法是使用某种反应性机制，如Emitter处理器，并返回它。使用此代码，您的代码将类似于：

@Service
class StudentsQueueWriter {
    val students: EmitterProcessor<Student> = EmitterProcessor.create()
    @Bean
    fun studentsChannel(): Supplier<Flux<Student>> {
        return Supplier { students }
    }
}

其用法可能类似于：

class MyClass(val studentsQueueWriter: StudentsQueueWriter) {
    fun newStudent() {
        studentsQueueWriter.students.onNext(Student("Adam"))
    }
}