代码之家 › 专栏 › 技术社区 › kleash

Spring集成JAVA DSL HTTP没有在超时错误内接收到应答

spring-integration-http spring-integration-dsl spring-integration java

1

kleash · 技术社区 · 8 年前

我使用的是SpringIntegration5.0.6。我已经浏览了它的文档并创建了以下代码,这些代码在HTTP端点上监听并发布到卡夫卡主题。

一切都很好,我也收到了这个主题的信息。但是在HTTP客户机上没有发送回复,而是给出“超时内没有收到回复”。

如何以下面的代码向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));
    }

谢谢你的帮助。

1 回复 | 直到 8 年前

1

Artem Bilan 8 年前

你的问题是你使用单向 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(产品工厂) 第二个可以是上面提到的 .enrichHeaders() . 如果您再不做任何操作,最后一个将把结果发送到 replyChannel 因此,头将到达HTTP响应。

在这个发布订阅场景中,您应该记住 payload 因为第二个订户将和你试图发送给卡夫卡的一样。