标签: spring-integration-amqp

和我们这里有类似的情况Spring Cloud Stream 和 RabbitMQ 健康检查

错误信息

2018-12-04 05:17:31.768 INFO 1 --- [nio-8080-exec-6] osarcCachingConnectionFactory ：尝试连接到：[localhost:5672]
2018-12-04 05:17:31.775 WARN 1 --- [nio-8080-exec-6] osbahealth.RabbitHealthIndicator：健康检查失败

org.springframework.amqp.AmqpConnectException：java.net.ConnectException：连接被拒绝（连接被拒绝）
    在 org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:62) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:387) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:586) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1436) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1417) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1393) ~[spring-rabbit-1.7.9.RELEASE.jar:na]
    在 org.springframework.boot.actuate.health.RabbitHealthIndicator.getVersion(RabbitHealthIndicator.java:49) ~[spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.health.RabbitHealthIndicator.doHealthCheck(RabbitHealthIndicator.java:45) ~[spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.health.Abs​​tractHealthIndicator.health(AbstractHealthIndicator.java:43) ~[spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.health.CompositeHealthIndicator.health(CompositeHealthIndicator.java:68) [spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.endpoint.HealthEndpoint.invoke(HealthEndpoint.java:86) [spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.endpoint.HealthEndpoint.invoke(HealthEndpoint.java:36) [spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.endpoint.mvc.AbstractEndpointMvcAdapter.invoke(AbstractEndpointMvcAdapter.java:56) [spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 org.springframework.boot.actuate.endpoint.mvc.EndpointMvcAdapter.invoke(EndpointMvcAdapter.java:42) [spring-boot-actuator-1.5.15.RELEASE.jar:1.5.15.RELEASE]
    在 sun.reflect.GeneratedMethodAccessor559.invoke(Unknown Source) ~[na:na]
    在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_171]
    在 java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_171]
    在 org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205) [spring-web-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    在 …

我开发了异步Spring Cloud Stream服务,我正在尝试开发一种边缘服务,它使用@MessagingGateway来提供对异步服务的同步访问.

我目前正在获得以下堆栈跟踪:

Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:355)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
... 47 common frames omitted

我的@MessagingGateway:

@EnableBinding(AccountChannels.class)
@MessagingGateway

public interface AccountService {
  @Gateway(requestChannel = AccountChannels.CREATE_ACCOUNT_REQUEST,replyChannel = AccountChannels.ACCOUNT_CREATED, replyTimeout = 60000, requestTimeout = 60000)
  Account createAccount(@Payload Account account, @Header("Authorization") String authorization);
}

如果我通过@StreamListener在回复频道上使用该消息,它可以正常工作:

  @HystrixCommand(commandKey = "acounts-edge:accountCreated", fallbackMethod = "accountCreatedFallback", commandProperties = {@HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE")}, ignoreExceptions = {ClientException.class})
  @StreamListener(AccountChannels.ACCOUNT_CREATED)
  public void accountCreated(Account account, @Header(name …

我有一个使用 Spring Cloud Stream Rabbit 和 Eureka Discovery Client 的简单 Spring Boot 应用程序。该应用程序与 Eureka Server 一起工作得很好，并且通过 RabbitMQ 的消息传递也工作正常。但是如果我启动一个 Spring Boot Admin Server，应用程序就会开始记录健康检查失败：

2017-06-21 19:47:57.352  INFO 11416 --- [trap-executor-0] c.n.d.s.r.aws.ConfigClusterResolver      : Resolving eureka endpoints via configuration
2017-06-21 19:51:25.047  INFO 11416 --- [nio-8303-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring FrameworkServlet 'dispatcherServlet'
2017-06-21 19:51:25.047  INFO 11416 --- [nio-8303-exec-1] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization started
2017-06-21 19:51:25.069  INFO 11416 --- [nio-8303-exec-1] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization completed in 22 ms
2017-06-21 19:51:27.288  WARN 11416 --- [nio-8303-exec-1] …

我不确定为什么会出现异常

Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available

它只是一个简单的 IntegrationFlow，但不确定我在下面的代码中缺少什么。

  @Bean
  Exchange messageExchange() {
    return ExchangeBuilder
        .directExchange("attr")
        .durable(true)
        .build();
  }

  @Bean
  Queue queue() {
    return QueueBuilder
        .durable("attr_queue")
        .build();
  }

  @Bean
  Binding binding() {
    return BindingBuilder
        .bind(queue())
        .to(messageExchange())
        .with("attr_queue")
        .noargs();
  }

  @Bean
  IntegrationFlow deltaFlow(ConnectionFactory connectionFactory) {
    return IntegrationFlows.from(Amqp
        .inboundAdapter(connectionFactory, queue()))
        .handle(String.class, (payload, headers) -> {
          if (payload.isEmpty()) {
            log.info("Payload empty");
          } else {
            log.info("Payload : " + payload);
          }
          return payload;
        })
        .get();
  }

我试图接触 Spring Integration，但不确定为什么我会收到此异常。我要做的就是使用 读取队列并将其inboundAdapter …