spring-boot - Turbine AMQP 未从服务接收到 hystrix.stream

Question

我正在尝试实现 Turbine AMQP 以将来自多个服务的所有 hystrix.stream 合并为一个并将其显示在 hystrix 仪表板中。

我的服务中的 Hystrix.stream 看起来不错：-

我在客户端服务的流中看到的示例：- localhost:4444/hystrix.stream

ping: 

data: {"type":"HystrixCommand","name":"getUserAssociations","group":"UserAssociationsDAO","currentTime":1462220674908,"isCircuitBreakerOpen":false,"errorPercentage":0,"errorCount":0,"requestCount":0,"rollingCountBadRequests":0,"rollingCountCollapsedRequests":0,"rollingCountEmit":0,"rollingCountExceptionsThrown":0,"rollingCountFailure":0,"rollingCountFallbackEmit":0,"rollingCountFallbackFailure":0,"rollingCountFallbackMissing":0,"rollingCountFallbackRejection":0,"rollingCountFallbackSuccess":0,"rollingCountResponsesFromCache":0,"rollingCountSemaphoreRejected":0,"rollingCountShortCircuited":0,"rollingCountSuccess":0,"rollingCountThreadPoolRejected":0,"rollingCountTimeout":0,"currentConcurrentExecutionCount":0,"rollingMaxConcurrentExecutionCount":0,"latencyExecute_mean":0,"latencyExecute":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"latencyTotal_mean":0,"latencyTotal":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"propertyValue_circuitBreakerRequestVolumeThreshold":20,"propertyValue_circuitBreakerSleepWindowInMilliseconds":5000,"propertyValue_circuitBreakerErrorThresholdPercentage":50,"propertyValue_circuitBreakerForceOpen":false,"propertyValue_circuitBreakerForceClosed":false,"propertyValue_circuitBreakerEnabled":true,"propertyValue_executionIsolationStrategy":"SEMAPHORE","propertyValue_executionIsolationThreadTimeoutInMilliseconds":60000,"propertyValue_executionTimeoutInMilliseconds":60000,"propertyValue_executionIsolationThreadInterruptOnTimeout":true,"propertyValue_executionIsolationThreadPoolKeyOverride":null,"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_metricsRollingStatisticalWindowInMilliseconds":10000,"propertyValue_requestCacheEnabled":true,"propertyValue_requestLogEnabled":true,"reportingHosts":1,"threadPool":"UserAssociationsDAO"}

data: {"type":"HystrixCommand","name":"getUserAssociation","group":"UserAssociationsDAO","currentTime":1462220674909,"isCircuitBreakerOpen":false,"errorPercentage":0,"errorCount":0,"requestCount":0,"rollingCountBadRequests":0,"rollingCountCollapsedRequests":0,"rollingCountEmit":0,"rollingCountExceptionsThrown":0,"rollingCountFailure":0,"rollingCountFallbackEmit":0,"rollingCountFallbackFailure":0,"rollingCountFallbackMissing":0,"rollingCountFallbackRejection":0,"rollingCountFallbackSuccess":0,"rollingCountResponsesFromCache":0,"rollingCountSemaphoreRejected":0,"rollingCountShortCircuited":0,"rollingCountSuccess":0,"rollingCountThreadPoolRejected":0,"rollingCountTimeout":0,"currentConcurrentExecutionCount":0,"rollingMaxConcurrentExecutionCount":0,"latencyExecute_mean":0,"latencyExecute":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"latencyTotal_mean":0,"latencyTotal":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"propertyValue_circuitBreakerRequestVolumeThreshold":20,"propertyValue_circuitBreakerSleepWindowInMilliseconds":5000,"propertyValue_circuitBreakerErrorThresholdPercentage":50,"propertyValue_circuitBreakerForceOpen":false,"propertyValue_circuitBreakerForceClosed":false,"propertyValue_circuitBreakerEnabled":true,"propertyValue_executionIsolationStrategy":"SEMAPHORE","propertyValue_executionIsolationThreadTimeoutInMilliseconds":60000,"propertyValue_executionTimeoutInMilliseconds":60000,"propertyValue_executionIsolationThreadInterruptOnTimeout":true,"propertyValue_executionIsolationThreadPoolKeyOverride":null,"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_metricsRollingStatisticalWindowInMilliseconds":10000,"propertyValue_requestCacheEnabled":true,"propertyValue_requestLogEnabled":true,"reportingHosts":1,"threadPool":"UserAssociationsDAO"}

来自客户端的 Gradle 依赖项：-

dependencyManagement {
    imports { 
        mavenBom "org.springframework.cloud:spring-cloud-starter-parent:Brixton.M5" 
    }
}

dependencies {
    compile         ("org.springframework.cloud:spring-cloud-starter-hystrix:1.0.6.RELEASE")
    compile 'org.springframework.cloud:spring-cloud-netflix-hystrix-stream:1.1.0.M2'
    compile ('org.springframework.cloud:spring-cloud-starter-bus-amqp:1.0.6.RELEASE')
    compile         ("org.springframework.boot:spring-boot-starter-actuator")
    compile         ("org.springframework.cloud:spring-cloud-starter-eureka")
    compile         ("org.springframework.cloud:spring-cloud-config-client")

    compile         ("org.codehaus.sonar-plugins.java:sonar-jacoco-plugin:2.3")
    compile         ("org.springframework.boot:spring-boot-starter-web")
    compile         ("org.codehaus.jackson:jackson-mapper-asl:1.9.2")
    compile         ('com.netflix.hystrix:hystrix-javanica:1.5.2')
    compile         ("mysql:mysql-connector-java:5.1.34")
    compile         ("org.springframework:spring-webmvc:4.2.2.RELEASE")
    compile         ("org.springframework.boot:spring-boot-starter-data-jpa")
    providedRuntime ("org.springframework.boot:spring-boot-starter-tomcat")
    testCompile     ("org.springframework.boot:spring-boot-starter-test") 
    querydslapt     ("org.hibernate:hibernate-jpamodelgen:5.0.5.Final")
        compile 'org.springframework.cloud:spring-cloud-starter-sleuth:1.0.0.M1'
    compile 'org.springframework.cloud:spring-cloud-sleuth-core:1.0.0.M1' 

}

所以我确实有一个正在运行的 Rabbit MQ 服务器和 TurboAMQP 项目。

我的turtleAMQP 注释类：-

@SpringBootApplication
@Configuration
@EnableAutoConfiguration
@EnableTurbineAmqp
@EnableDiscoveryClient
public class TurbineApplication {

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

应用程序.yml

info:
  component: Turbine

endpoints: 
  restart:
    enabled: true
  shutdown:
    enabled: true

turbine:
  amqp:
    port: 8989
management:
  port: 8990

spring:
 rabbitmq:
   host: localhost
   port: 5672
   username: guest
   password: guest


eureka:
 client:
   serviceUrl:
     defaultZone: ${eurekaurl:http://localhost:8761/eureka/}

gradle 对 Turbine 的依赖：-

dependencyManagement {
    imports { 
        mavenBom "org.springframework.cloud:spring-cloud-starter-parent:Brixton.M5" 
    }
}
dependencies {
    compile('org.springframework.boot:spring-boot-starter-actuator')
    compile('org.springframework.boot:spring-boot-actuator-docs')
    compile('org.springframework.cloud:spring-cloud-starter-config')
    compile('org.springframework.cloud:spring-cloud-config-client')
    compile('org.springframework.cloud:spring-cloud-starter-turbine-amqp:1.0.6.RELEASE')
    compile('org.springframework.cloud:spring-cloud-starter-eureka')
    compile('org.springframework.boot:spring-boot-starter-logging')
    compile("org.springframework.boot:spring-boot-starter-web")
    testCompile('org.springframework.boot:spring-boot-starter-test') 
}

当我尝试点击http://localhost:8989/turbine.stream时，它只会显示 ping，这就是我在尝试放置 turbione.stream 时在日志中看到的内容，尽管它在 hystrix 仪表板中不起作用。

2016-05-02 13:55:05.865  INFO 9028 --- [o-eventloop-3-4] o.s.c.n.t.amqp.TurbineAmqpConfiguration  : SSE Request Received
2016-05-02 13:55:05.906 ERROR 9028 --- [o-eventloop-3-3] i.r.netty.server.DefaultErrorHandler     : Unexpected error in RxNetty.

java.io.IOException: An established connection was aborted by the software in your host machine
    at sun.nio.ch.SocketDispatcher.read0(Native Method) ~[na:1.8.0_66]
    at sun.nio.ch.SocketDispatcher.read(Unknown Source) ~[na:1.8.0_66]
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(Unknown Source) ~[na:1.8.0_66]
    at sun.nio.ch.IOUtil.read(Unknown Source) ~[na:1.8.0_66]
    at sun.nio.ch.SocketChannelImpl.read(Unknown Source) ~[na:1.8.0_66]
    at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:311) ~[netty-buffer-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:881) ~[netty-buffer-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:241) ~[netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:119) ~[netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511) [netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468) [netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382) [netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354) [netty-transport-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111) [netty-common-4.0.27.Final.jar:4.0.27.Final]
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137) [netty-common-4.0.27.Final.jar:4.0.27.Final]
    at java.lang.Thread.run(Unknown Source) [na:1.8.0_66]

我在这里做错了什么来让turtle.stream工作。

我的涡轮机知道 RabbitMQ，但客户端服务不知道。我是否需要告诉服务我也知道涡轮机知道来自尤里卡的所有 hystrix.stream(s)，所以涡轮机负责在 Rabbit MQ 中对 hystrix.stream 进行排队。

感谢您提供任何帮助我已经被困了一段时间了，并且没有很好的例子可以提供帮助。