当前位置: 首页 > 知识库问答 >
问题:

涡轮AMQP不接收Hystrix流

唐俊英
2023-03-14
spring:
  application:
    name: policy-service
eureka:
  client:
    serviceUrl:
      defaultZone: http://localhost:8761/eureka/
spring:
  rabbitmq:
    addresses: ${vcap.services.${PREFIX:}rabbitmq.credentials.uri:amqp://${RABBITMQ_HOST:localhost}:${RABBITMQ_PORT:5672}}

启动日志的尾端如下所示:

2015-09-14 16:31:13.030  INFO 52844 --- [           main] com.netflix.discovery.DiscoveryClient    : Starting heartbeat executor: renew interval is: 30
2015-09-14 16:31:13.047  INFO 52844 --- [           main] c.n.e.EurekaDiscoveryClientConfiguration : Registering application policy-service with eureka with status UP
2015-09-14 16:31:13.194  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'policy-service:8088.errorChannel' has 1 subscriber(s).
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started _org.springframework.integration.errorLogger
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {filter} as a subscriber to the 'cloudBusOutboundFlow.channel#0' channel
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-    service:8088.cloudBusOutboundFlow.channel#0' has 1 subscriber(s).
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#0
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {filter} as a subscriber to the 'cloudBusInboundChannel' channel
2015-09-14 16:31:13.195  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.cloudBusInboundChannel' has 1 subscriber(s).
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#1
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {message-handler} as a subscriber to the 'cloudBusInboundFlow.channel#0' channel
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.cloudBusInboundFlow.channel#0' has 1 subscriber(s).
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#2
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter} as a subscriber to the 'cloudBusWiretapChannel' channel
2015-09-14 16:31:13.196  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.cloudBusWiretapChannel' has 1 subscriber(s).
2015-09-14 16:31:13.197  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#3
2015-09-14 16:31:13.197  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {amqp:outbound-channel-adapter} as a subscriber to the 'cloudBusOutboundChannel' channel
2015-09-14 16:31:13.197  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.cloudBusOutboundChannel' has 1 subscriber(s).
2015-09-14 16:31:13.197  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#4
2015-09-14 16:31:13.198  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {bridge} as a subscriber to the 'cloudBusAmqpInboundFlow.channel#0' channel
2015-09-14 16:31:13.198  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.cloudBusAmqpInboundFlow.channel#0' has 1 subscriber(s).
2015-09-14 16:31:13.198  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#5
2015-09-14 16:31:13.198  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {amqp:outbound-channel-adapter} as a subscriber to the 'hystrixStream' channel
2015-09-14 16:31:13.199  INFO 52844 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'policy-service:8088.hystrixStream' has 1 subscriber(s).
2015-09-14 16:31:13.199  INFO 52844 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#6
2015-09-14 16:31:13.219  INFO 52844 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 1073741823
2015-09-14 16:31:13.219  INFO 52844 --- [           main] ApplicationEventListeningMessageProducer : started org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer#0
2015-09-14 16:31:13.555  INFO 52844 --- [cTaskExecutor-1] o.s.amqp.rabbit.core.RabbitAdmin         : Auto-declaring a non-durable, auto-delete, or exclusive Queue (4640c1c8-ff8f-45d7-8426-19d1b7a4cdb0) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2015-09-14 16:31:13.572  INFO 52844 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter#0
2015-09-14 16:31:13.573  INFO 52844 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 2147483647
2015-09-14 16:31:13.576  INFO 52844 --- [           main] c.n.h.c.m.e.HystrixMetricsPoller         : Starting HystrixMetricsPoller
2015-09-14 16:31:13.609  INFO 52844 --- [           main] ration$HystrixMetricsPollerConfiguration : Starting poller
2015-09-14 16:31:13.803  INFO 52844 --- [           main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 8088 (http)
2015-09-14 16:31:13.805  INFO 52844 --- [           main] com.ml.springboot.PolicyService     : Started PolicyService in 22.544 seconds (JVM running for 23.564)

因此,PolicyService似乎成功连接到message Broker。

Turbine AMQP服务器的日志结束:

2015-09-14 16:58:05.887  INFO 51944 --- [           main] i.reactivex.netty.server.AbstractServer  : Rx server started at port: 8989
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'bootstrap:-1.errorChannel' has 1 subscriber(s).
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started _org.springframework.integration.errorLogger
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {bridge} as a subscriber to the 'hystrixStreamAggregatorInboundFlow.channel#0' channel
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'bootstrap:-1.hystrixStreamAggregatorInboundFlow.channel#0' has 1 subscriber(s).
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.config.ConsumerEndpointFactoryBean#0
2015-09-14 16:58:05.991  INFO 51944 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 1073741823
2015-09-14 16:58:06.238  INFO 51944 --- [cTaskExecutor-1] o.s.amqp.rabbit.core.RabbitAdmin         : Auto-declaring a non-durable, auto-delete, or exclusive Queue (spring.cloud.hystrix.stream) durable:false, auto-delete:false, exclusive:false. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2015-09-14 16:58:06.289  INFO 51944 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter#0
2015-09-14 16:58:06.290  INFO 51944 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 2147483647
2015-09-14 16:58:06.434  INFO 51944 --- [           main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): -1 (http)
package com.turbine.amqp;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.netflix.turbine.amqp.EnableTurbineAmqp;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableAutoConfiguration
@EnableTurbineAmqp
@EnableDiscoveryClient
public class TurbineAmqpApplication {

    public static void main(String[] args) {
        SpringApplication.run(TurbineAmqpApplication.class, args);
    }
}
server:
  port: 8989
spring:
  rabbitmq:
addresses: ${vcap.services.${PREFIX:}rabbitmq.credentials.uri:amqp://${RABBITMQ_HOST:localhost}:${RABBITMQ_PORT:5672}}
2015-09-15 08:54:37.960  INFO 83480 --- [o-eventloop-3-1] o.s.c.n.t.amqp.TurbineAmqpConfiguration  : SSE Request Received
2015-09-15 08:54:38.025  INFO 83480 --- [o-eventloop-3-1] o.s.c.n.t.amqp.TurbineAmqpConfiguration  : Starting aggregation

编辑:下面的异常是当我停止收听涡轮流时抛出的,而不是当我尝试用仪表板收听时抛出的。

2015-09-15 08:56:47.934  INFO 83480 --- [o-eventloop-3-3] o.s.c.n.t.amqp.TurbineAmqpConfiguration  : SSE Request Received
2015-09-15 08:56:47.946  WARN 83480 --- [o-eventloop-3-3] io.netty.channel.DefaultChannelPipeline  : An exception was thrown by a user handler's exceptionCaught() method while handling the following exception:

java.lang.NoSuchMethodError: rx.Observable.collect(Lrx/functions/Func0;Lrx/functions/Action2;)Lrx/Observable;
    at com.netflix.turbine.aggregator.StreamAggregator.lambda$null$36(StreamAggregator.java:89)
    at rx.internal.operators.OnSubscribeMulticastSelector.call(OnSubscribeMulticastSelector.java:60)
    at rx.internal.operators.OnSubscribeMulticastSelector.call(OnSubscribeMulticastSelector.java:40)
    at rx.Observable.unsafeSubscribe(Observable.java:8591)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.handleNewSource(OperatorMerge.java:190)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:160)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:96)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
    at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.onNext(OperatorGroupBy.java:173)
    at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java:224)
    at rx.subjects.PublishSubject.onNext(PublishSubject.java:101)
    at org.springframework.cloud.netflix.turbine.amqp.Aggregator.handle(Aggregator.java:53)
    at sun.reflect.GeneratedMethodAccessor63.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:497)
    at org.springframework.expression.spel.support.ReflectiveMethodExecutor.execute(ReflectiveMethodExecutor.java:112)
    at org.springframework.expression.spel.ast.MethodReference.getValueInternal(MethodReference.java:102)
    at org.springframework.expression.spel.ast.MethodReference.access$000(MethodReference.java:49)
    at org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:342)
    at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:88)
    at org.springframework.expression.spel.ast.SpelNodeImpl.getTypedValue(SpelNodeImpl.java:131)
    at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:330)
    at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:164)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:276)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:142)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:75)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:71)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:99)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:277)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:239)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:95)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:248)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:171)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:119)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:105)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:277)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:239)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:95)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:101)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$400(AmqpInboundChannelAdapter.java:45)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$1.onMessage(AmqpInboundChannelAdapter.java:93)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:756)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:679)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:82)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:167)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1241)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:660)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1005)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:989)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:82)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1103)
    at java.lang.Thread.run(Thread.java:745)
Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: GroupedObservable.class
    at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:56)
    ... 58 common frames omitted

我对Turbane-AMQP的依赖关系如下:

dependencies {
    compile('org.springframework.cloud:spring-cloud-starter-turbine-amqp:1.0.3.RELEASE')
    compile 'org.springframework.boot:spring-boot-starter-web:1.2.5.RELEASE'
    compile 'org.springframework.boot:spring-boot-starter-actuator:1.2.5.RELEASE'

    testCompile("org.springframework.boot:spring-boot-starter-test") 
}

dependencyManagement {
    imports { 
         mavenBom 'org.springframework.cloud:spring-cloud-starter-parent:1.0.2.RELEASE'
    }
}

共有1个答案

冯福
2023-03-14

很难找到解决办法。

使用Spring Cloud2.1.4.Release,我也面临类似的问题。

主要原因是rabbitMQ中的名称:spring-cloud-netflix-hystrix-stream和spring-cloud-starter-netflix-turbin-stream不兼容。

 类似资料:
  • 我正在使用turbine来汇总Hystrix的数据。当我尝试用hystrix(localhost:23002/hystrix.strean)获取数据时,一切都很顺利。然而,当我尝试使用turbine获取数据时,它显示什么也没有。这是我的配置。 我可以通过URL:localhost:23002/hystrix.stream获取数据我的目标是通过URL localhost:23111/turbine.

  • 我有一个spring云应用程序充当涡轮服务器和Hystrix仪表盘。它有一个Eureka客户端,仪表板出现了,可以看到一个单独的Hystrix流,但我没有得到任何涡轮流。 这是application.yml: 如果我在Turbin/Dashboard应用程序启动时查看日志,它会发现我的Hystrix服务正常: 如果我撞上涡轮流 我得到: 看起来涡轮程序发现了两个项目-通过Eureka的服务和他们的

  • 我的设置是spring boot cloud,使用netflix library,我设法让Turbine从一个服务中聚合Hystrix度量。然而,当我添加更多的服务时,我看不到它们。 这是我的设置(也上传到github中,网址是:Project On github 服务2: 优惠券服务: 尤里卡应用程序服务:

  • 我正在更新以使用consul的涡轮机,从文档中我看到本节: 群集配置和 appConfig 部分必须匹配,因此将服务 ID 的逗号分隔列表放入单独的配置属性中很有用。 那么,我应该在申请表上写些什么呢?这是实际的服务名称吗? 不清楚。

  • 这里是我的application.yml的相关部分 从文档中,我发现应该创建一个SpringAMQPMessageSource bean: 如果我从rabbitmq管理面板向队列发送消息,我会看到日志: SpringAMQPMessageSource的JavaDoc是这样说的: 但到现在我都找不到在哪里注册,也找不到怎么注册。 配置中的Axon.EventHandling条目和聚合中的@Proce

  • 目标:使用python从EventHub接收消息。 设置: Python 3.5 AMQP 1.0 在新门户中创建的EventHub。 Ubuntu 16.04和Mac OS X。 质子==0.8.8和python-qpid-质子==0.17.0 几天来,我一直在尝试连接AMQP和Azure EventHub。我可以用Java客户端很好地做到这一点(作为确保EventHub工作的测试),但我需要能