这是通过 Spring Integration Flow 处理结果的正确方法吗？

Question

我目前正在做一个个人项目 - 我需要我的 Spring 应用程序从 EMQX（MQTT 服务器）获取查询并查询其数据以获得相应的结果，然后将结果推送到带有查询 UUID 的主题。

这是有效的——经过数小时了解 Spring Integration 框架的工作原理。但我认为处理程序使用“块”的方式是不正确的 - 并且不符合集成流程应该运行的方式。虽然这项工作有效，但出于对工作的尊重，我确实想确保它被正确完成，并避免将来出现问题。

下面的代码 sn-p 应该足以理解我想要实现的目标 - 以及潜在问题所在。

    @Bean
fun mqttInFlow() : Publisher<Message<String>> {
    return IntegrationFlows.from(inbound())
        .handle<String> { payload, headers ->
            val emotionalOutput: EmotionalOutput = gson.fromJson(payload, EmotionalOutput::class.java)
            emotionalPrintService.populateEmotionalOutput(emotionalOutput).map {
                MessageBuilder.withPayload(gson.toJson(it))
                    .copyHeaders(headers)
                    .setHeader(MqttHeaders.TOPIC, "query/" + it.query_uuid).build()
            }.block()
        }
        .channel(outgoingChannel())
        .toReactivePublisher()
}

编辑 - 感谢您的建议 - 这是我理解的 Kotlin DSL 解决方案的潜在编辑 - 现在产生错误 - 抱怨输出通道或回复通道不可用 - 除了 this功能已更改。

    @Bean
fun newMqttInFlow() =
    integrationFlow (inbound()) {
       wireTap {
            handle<String> { payload, headers ->
                gson.fromJson<EmotionalOutput>(payload, EmotionalOutput::class.java).let { emotionalOutput ->
                    emotionalPrintService.populateEmotionalOutput(emotionalOutput).map { populatedEmotionalOutput ->
                        MessageBuilder.withPayload(gson.toJson(populatedEmotionalOutput))
                            .copyHeaders(headers)
                            .setHeader(MqttHeaders.TOPIC, populatedEmotionalOutput.query_uuid)
                    }
                }
            }
        }
        channel("outgoingChannel")
    }

例外是：

例外是 org.springframework.messaging.core.DestinationResolutionException：没有可用的输出通道或回复通道标头

虽然我有多年的 Java 经验 - 这种方法是新的 - 所以非常感谢您的帮助。值得赞赏。如果整个课程有用 - 我可以发布。

编辑

这是配置文件 - 可以更好地了解可能导致此次要错误的原因 -

021-03-28 21:59:48.008 错误 84492 --- [T 调用：divnrin] os.integration.handler.LoggingHandler：org.springframework.messaging.MessageHandlingException：消息处理程序 [bean 'mqttOutbound' 发生错误；定义在：'类路径资源 [io/divnr/appserver/configuration/MQTTConfiguration.class]';来源：'org.springframework.core.type.classreading.SimpleMethodMetadata@4a9419d7']；嵌套异常是 java.lang.IllegalArgumentException：此默认转换器只能处理 'byte[]' 或 'String' 有效负载；考虑在流定义中添加转换器，或提供 BytesMessageMapper，或为 reactor.core.publisher.MonoMapFuseable 有效负载子类化此转换器，failedMessage=GenericMessage [payload=MonoMapFuseable, headers={mqtt_receivedRetained=false, mqtt_id=0, mqtt_duplicate=false , id=c5a75283-c0fe-ebac-4168-dabddd989da9, mqtt_receivedTopic=source/d9e50e8f-67e0-4505-7ca2-4d05b1242207, mqtt_receivedQos=0, 时间戳=1616961588004}] 在 org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:192) 在 org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:65) 在

这里提供了完整的课程。

@Configuration
@EnableIntegration
@IntegrationComponentScan
class MQTTConfiguration(val emotionalPrintService: EmotionalPrintService,
                    val gson: Gson,
                    val applicationConfiguration: ApplicationConfiguration) {

@Bean
fun mqttServiceFactory() : MqttPahoClientFactory {
    return DefaultMqttPahoClientFactory().apply {
        connectionOptions = MqttConnectOptions().apply {
            serverURIs = arrayOf<String>(applicationConfiguration.mqttServerAddress)
        }
    }
}

@Bean
fun newMqttInFlow() =
    integrationFlow (inbound()) {
        handle<String> { payload, headers ->
            gson.fromJson<EmotionalOutput>(payload, EmotionalOutput::class.java).let { emotionalOutput ->
                emotionalPrintService.populateEmotionalOutput(emotionalOutput).map { populatedEmotionalOutput ->
                    MessageBuilder.withPayload(gson.toJson(populatedEmotionalOutput))
                        .copyHeaders(headers)
                        .setHeader(MqttHeaders.TOPIC, populatedEmotionalOutput.query_uuid).build()
                }
            }
        }
        channel(outgoingChannel())
    }

@Bean
@ServiceActivator(requiresReply = "false", inputChannel = "outgoingChannel")
fun mqttOutbound(): MessageHandler {
    val messageHandler = MqttPahoMessageHandler("divnrout", mqttServiceFactory())
    messageHandler.setAsync(true)
    return messageHandler
}

@Bean
fun outgoingChannel() : FluxMessageChannel {
    return FluxMessageChannel()
}

@Bean
fun inbound(): MessageProducerSupport {
    return MqttPahoMessageDrivenChannelAdapter("divnrin", mqttServiceFactory(),
        "source/" + applicationConfiguration.sourceUuid).apply {
        setConverter(DefaultPahoMessageConverter())
        setQos(1)
    }
}
}

Answer 1

您确实不需要block() 在您的handle() 末尾。您只需从该emotionalPrintService.populateEmotionalOutput() 返回Mono，框架将为您提供有关正确订阅和背压处理的信息。

您还需要将outgoingChannel() 设为FluxMessageChannel。 在文档中查看更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/reactive-streams.html#reactive-streams

另外考虑将您的 IntegrationFlow 解决方案移至正确的 Kotlin DSL：https://docs.spring.io/spring-integration/docs/current/reference/html/kotlin-dsl.html#kotlin-dsl

另外：当它是流结束时的FluxMessageChannel 时，没有理由担心toReactivePublisher() - FluxMessageChannel 本身就是Publisher<Message<?>>。

更新

问题出在这里：

handle<String>( { payload, headers ->
        gson.fromJson<EmotionalOutput>(payload, EmotionalOutput::class.java).let { emotionalOutput ->
            emotionalPrintService.populateEmotionalOutput(emotionalOutput).map { populatedEmotionalOutput ->
                MessageBuilder.withPayload(gson.toJson(populatedEmotionalOutput))
                    .copyHeaders(headers)
                    .setHeader(MqttHeaders.TOPIC, populatedEmotionalOutput.query_uuid).build()
            }
        }
    }) { async(true) }

查看async(true) 选项。不幸的是，在当前版本中，默认情况下我们不允许它以响应方式处理响应式回复。你必须说你想在这个端点成为异步。因此，您的Publisher 回复和FluxMessageChannel 作为输出将起到正确的作用。