Spring Integration DSL을 사용하여 rabbitMQ 응답 큐에서 응답이 올 때 HTTP 요청 / 응답을 구현하는 방법은 무엇입니까?

월리 리우

Spring Integration DSL에서 별도의 RabbitMQ 큐를 사용하여 HTTP 요청 / 응답을 구현하려고합니다. amqp queue에 대한 Spring IntegrationFlow http 요청 과 유사합니다 . 차이점은 원래 http 호출자에게 응답을 원한다는 것입니다. 테스트 http post 메시지가 성공적으로 요청 큐에 전달되고 응답 큐로 변환 (대문자로)되는 것을 볼 수 있습니다. 메시지는 응답 대기열에서도 사용되었지만 호출자 (http : // localhost : 8080 / Tunner)에게 다시 반환되지 않았습니다. 결국 500 오류와 함께 호출 시간이 초과되었습니다. 나는 이것에 익숙하지 않아서 내가 완전히 놓친 것이있을 수 있습니다. 누군가 제안을 할 수 있습니까? 코드는 다음과 같습니다.

public class TunnelApplication
{
    public static void main(String[] args)
    {
        SpringApplication.run(TunnelApplication.class, args);
    }

    @Value("${outboundQueue}")
    private String outboundQueue;

    @Value("${inboundQueue}")
    private String inboundQueue;

    private ConnectionFactory rabbitConnectionFactory;

    @Autowired
    public TunnelApplication(ConnectionFactory factory) {
        rabbitConnectionFactory = factory;
    }

    @Bean
    public Queue targetQueue()
    {
        return new Queue(outboundQueue, true, false, true);
    }

    @Bean
    public Queue requestQueue()
    {
        return new Queue(inboundQueue, true, false, true);
    }

    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter()
    {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate amqpTemplate()
    {
        RabbitTemplate result = new RabbitTemplate(rabbitConnectionFactory);
        result.setMessageConverter(jsonMessageConverter());
        result.setDefaultReceiveQueue(outboundQueue);
        //result.setReplyAddress(outboundQueue);
        result.setReplyTimeout(60000);
        return result;
    }

    @Bean
    public IntegrationFlow sendReceiveFlow(RabbitTemplate amqpTemplate) {
        return IntegrationFlows
                .from(Http.inboundGateway("/tunnel"))
                .handle(Amqp.outboundGateway(amqpTemplate)
                        .routingKey(inboundQueue)
                        .returnChannel(amqpOutboundChannel()))
                .log()
                .bridge(null)
                .get();
    }

    @Bean
    public IntegrationFlow rabbitToWeb(RabbitTemplate amqpTemplate, ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Amqp.inboundGateway(connectionFactory, requestQueue()))
                .transform(String.class, String::toUpperCase)
                .log()
                .handle(Amqp.outboundGateway(amqpTemplate).routingKey(outboundQueue))
                .log()
                .bridge(null)
                .get();
    }

    @Bean
    public IntegrationFlow replyBackToHttp(RabbitTemplate amqpTemplate, ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Amqp.inboundGateway(connectionFactory, targetQueue()))
                .handle(Http.outboundGateway("http://localhost:8080/tunnel")
                       .expectedResponseType(String.class))
                .log()
                .bridge(null)
                .channel(amqpOutboundChannel())
                .get();
    }

    @Bean
    public MessageChannel amqpOutboundChannel() {
    return new DirectChannel();
}

또한 동료가 다음 코드를 시도했지만 응답을받지 못했습니다.

@Configuration
@EnableIntegration
public class FlowConfig {

 

   @Value("${routingKey}")
   private String routingKey;

 

   @Value("${rabbitSinkChannel}")
   private String rabbitSinkChannel;

 

   @Bean
   public MessageChannel rabbitSinkChannel(ConnectionFactory connectionFactory) {
      return
         Amqp
         .channel(rabbitSinkChannel, connectionFactory)
         .get();
   }

 

   @Bean
   public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
      return new RabbitTemplate(connectionFactory);
   }

 

   @Bean
   public IntegrationFlow httpFlow(RabbitTemplate rabbitTemplate, ConnectionFactory connectionFactory) {
      MessageChannel rabbitSinkChannel = rabbitSinkChannel(connectionFactory);

 

      return IntegrationFlows
         .from(
            Http.inboundGateway("/sendreceive")
         )
         .handle(
            Amqp.outboundGateway(rabbitTemplate)
               .routingKey(routingKey)
               .returnChannel(rabbitSinkChannel)
         )
         .channel(rabbitSinkChannel) // or .handle? if so, what?

 

         .get();
   }
}
월리 리우

다음 업데이트가 작동합니다 (repliesBackToHttp () 메서드도 제거했습니다).

@Bean
public AmqpTemplate amqpTemplate()
{
    RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
    rabbitTemplate.setMessageConverter(jsonMessageConverter());
    //result.setDefaultReceiveQueue(outboundQueue);
    rabbitTemplate.setReplyAddress(outboundQueue);
    rabbitTemplate.setReplyTimeout(60000);
    rabbitTemplate.setUseDirectReplyToContainer(false);
    return rabbitTemplate;
}

@Bean
public SimpleMessageListenerContainer replyListenerContainer() {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(rabbitConnectionFactory);
    container.setQueues(replyQueue());
    container.setMessageListener((MessageListener) amqpTemplate());
    return container;
}

이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.

침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제

에서 수정
0

몇 마디 만하겠습니다

0리뷰
로그인참여 후 검토

관련 기사

Related 관련 기사

뜨겁다태그

보관