Spring集成JavaDSL HTTP在超时错误内没有收到回复

Question

我正在使用Spring集成5.0.6。我已经浏览了它的文档并创建了以下代码，这些代码监听HTTPendpoint并发布到kafka主题。

一切都很好，我也收到了主题信息。但在HTTP客户端没有发送回复，它给出“在超时内没有收到回复”。

如何在下面的代码中向超文本传输协议调用者发送回复：

@Bean
public DirectChannel replyChannel() {
    return new DirectChannel();
}

@Bean(name = "restInputFlow")
public IntegrationFlow send() {
    return IntegrationFlows
            .from(Http.inboundGateway("/push").requestMapping(m -> m.methods(HttpMethod.POST))
                    .requestPayloadType(String.class).replyChannel(replyChannel()))
            .transform(new Transformer())
            .handle(kafkaMessageHandler(producerFactory(), getKafkaSourceTopic()))
            .enrichHeaders(
                    c -> c.header(org.springframework.integration.http.HttpHeaders.STATUS_CODE, HttpStatus.CREATED))
            .get();
}

private KafkaProducerMessageHandlerSpec<GenericRecord, GenericRecord, ?> kafkaMessageHandler(
            ProducerFactory<GenericRecord, GenericRecord> producerFactory, String topic) {

        return Kafka.outboundChannelAdapter(producerFactory)
                .messageKey("key").headerMapper(mapper())
                .topicExpression("headers[kafka_topic] ?: '" + topic + "'")
                .configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
    }

谢谢你的帮助。

Answer 1

您使用单向Kafka.outboundChannelAdapter（producerFactory）的问题。这只是为了“发送和忘记”。

如果您有兴趣生成一些后续过程，或者需要回复HTTP请求，则应考虑使用：

/**
 * The {@link org.springframework.integration.channel.PublishSubscribeChannel} {@link #channel}
 * method specific implementation to allow the use of the 'subflow' subscriber capability.
 * @param publishSubscribeChannelConfigurer the {@link Consumer} to specify
 * {@link PublishSubscribeSpec} options including 'subflow' definition.
 * @return the current {@link IntegrationFlowDefinition}.
 */
public B publishSubscribeChannel(Consumer<PublishSubscribeSpec> publishSubscribeChannelConfigurer) {

在流定义中，您的第一个订阅者实际上是Kafka。outboundChannelAdapter（producerFactory）第二个可以是前面提到的。如果您不做更多操作，最后一个将把结果发送到replyChannel头中，因此将到达HTTP响应。

在这个发布-订阅场景中，您应该记住，第二个订阅者的负载将与您尝试发送给Kafka的负载相同。