Hystrix 監控數據聚合 Turbine【Finchley 版】

原文地址：https://windmt.com/2018/04/17/spring-cloud-6-turbine/

上一篇咱們介紹了使用 Hystrix Dashboard 來展現 Hystrix 用於熔斷的各項度量指標。經過 Hystrix Dashboard，咱們能夠方便的查看服務實例的綜合狀況，好比：服務調用次數、服務調用延遲等。可是僅經過 Hystrix Dashboard 咱們只能實現對服務當個實例的數據展示，在生產環境咱們的服務是確定須要作高可用的，那麼對於多實例的狀況，咱們就須要將這些度量指標數據進行聚合。下面，在本篇中，咱們就來介紹一下另一個工具：Turbine。

準備工做

在開始使用 Turbine 以前，咱們先回顧一下上一篇中實現的架構，以下圖所示：

其中，咱們構建的內容包括：

• eureka-server：服務註冊中心
• eureka-producer：服務提供者
• eureka-consumer-hystrix：使用 Feign 和 Hystrix 實現的服務消費者
• hystrix-dashboard：用於展現eureka-consumer-hystrix服務的 Hystrix 數據

建立 Turbine

下面，咱們將在上述架構基礎上，引入 Turbine 來對服務的 Hystrix 數據進行聚合展現。 這裏咱們將分別介紹兩種聚合方式。

經過 HTTP 收集聚合

建立一個標準的 Spring Boot 工程，命名爲：turbine。

POM 配置

在 pom.xml 中添加如下依賴

1 2 3 4

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-turbine</artifactId> </dependency>

啓動類

在啓動類上使用@EnableTurbine註解開啓 Turbine

1 2 3 4 5 6 7 8

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

配置文件

在 application.yml 加入 Eureka 和 Turbine 的相關配置

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

spring: application: name: turbine server: port: 8080 management: port: 8081 eureka: client: service-url: defaultZone: http://localhost:7000/eureka/ turbine: app-config: eureka-consumer-hystrix cluster-name-expression: new String("default") combine-host-port: true

參數說明

• turbine.app-config參數指定了須要收集監控信息的服務名；
• turbine.cluster-name-expression 參數指定了集羣名稱爲 default，當咱們服務數量很是多的時候，能夠啓動多個 Turbine 服務來構建不一樣的聚合集羣，而該參數能夠用來區分這些不一樣的聚合集羣，同時該參數值能夠在 Hystrix 儀表盤中用來定位不一樣的聚合集羣，只須要在 Hystrix Stream 的 URL 中經過 cluster 參數來指定；
• turbine.combine-host-port參數設置爲true，可讓同一主機上的服務經過主機名與端口號的組合來進行區分，默認狀況下會以 host 來區分不一樣的服務，這會使得在本地調試的時候，本機上的不一樣服務聚合成一個服務來統計。

注意：new String("default")這個必定要用 String 來包一下，不然啓動的時候會拋出異常：

1	org.springframework.expression.spel.SpelEvaluationException: EL1008E: Property or field 'default' cannot be found on object of type 'com.netflix.appinfo.InstanceInfo' - maybe not public or not valid?

測試

在完成了上面的內容構建以後，咱們來體驗一下 Turbine 對集羣的監控能力。分別啓動

• eureka-server
• eureka-producer
• eureka-consumer-hystrix
• turbine
• hystrix-dashboard

訪問 Hystrix Dashboard 並開啓對 http://localhost:8080/turbine.stream 的監控，這時候，咱們將看到針對服務 eureka-consumer-hystrix 的聚合監控數據。

此時的架構以下圖所示：

經過消息代理收集聚合

Spring Cloud 在封裝 Turbine 的時候，還實現了基於消息代理的收集實現。因此，咱們能夠將全部須要收集的監控信息都輸出到消息代理中，而後 Turbine 服務再從消息代理中異步的獲取這些監控信息，最後將這些監控信息聚合並輸出到 Hystrix Dashboard 中。經過引入消息代理，咱們的 Turbine 和 Hystrix Dashoard 實現的監控架構能夠改爲以下圖所示的結構：

從圖中咱們能夠看到，這裏多了一個重要元素：RabbitMQ。對於 RabbitMQ 的安裝與基本時候咱們能夠查看以前的 MQ 系列，這裏不作過多的說明。下面，咱們能夠來構建一個新的應用來實現基於消息代理的 Turbine 聚合服務。

關於經過 MQ 的聚合，在 Finchley.RC1 版本下有好多坑，好在最後能正常運行了。

Turbine Stream

UPDATED：2018-06-01
Finchley.RC2 已經出了，下邊提到的 bug 都已經被修復了，直接添加 @EnableTurbineStream 就能夠正常使用了。最新代碼實例請看：https://github.com/zhaoyibo/spring-cloud-study/tree/master/turbine-stream

須要注意一點的是，Turbine Stream 默認的端口已經從 8989 改成 8080 了。

UPDATED：2018-05-18
如下關於 Turbine Stream 的內容僅適用於 Finchley.RC1 版本。今天嘗試一下最新的 Finchley.BUILD-SNAPSHOT 發現 netty 的默認端口已經從 8989 改到 8080，而且須要依賴spring-cloud-starter-netflix-hystrix，由於目前 BUILD-SNAPSHOT 依舊有 bug 不肯定他們會怎麼改，我就暫時先不更新了。等到 RC2 release 的時候再來更新一發。

建立一個標準的 Spring Boot 工程，命名爲：turbine-stream-rabbitmq

POM

1 2 3 4 5 6 7 8

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-turbine-stream</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>

配置文件

1 2 3 4 5 6 7

spring: application: name: turbine-stream-rabbitmq eureka: client: service-url: defaultZone: http://localhost:7000/eureka/

啓動類

1 2 3 4 5 6 7 8 9 10 11 12 13 14

@SpringBootApplication @EnableTurbineStream public class TurbineStreamRabbitmqApplication { public static void main(String[] args) { SpringApplication.run(TurbineStreamRabbitmqApplication.class, args); } @Bean public ConfigurableCompositeMessageConverter integrationArgumentResolverMessageConverter(CompositeMessageConverterFactory factory) { return new ConfigurableCompositeMessageConverter(factory.getMessageConverterForAllRegistered().getConverters()); } }

改造服務調用者

以以前的 eureka-consumer-hystrix 項目爲基礎，在 pom.xml 里加入如下依賴

1 2 3 4 5 6 7 8

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-netflix-hystrix-stream</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>

再在啓動類上加上@EnableHystrix註解

1 2 3 4 5 6 7 8 9 10

@EnableHystrix @EnableFeignClients @SpringBootApplication public class EurekaConsumerHystrixApplication { public static void main(String[] args) { SpringApplication.run(EurekaConsumerHystrixApplication.class, args); } }

測試

分別啓動 eureka-consumer-hystrix、turbine-stream-rabbitmq 這兩個項目，而後在 RabbitMQ 的管理後臺能夠看到，自動建立了一個 Exchange 和 Queue

看到這仍是挺高興的，可是……

當訪問了一下 consumer 中的接口後，就開始了艱辛的爬坑路程……

遇到的坑

依賴

這個 Turbine Stream 以前應該是叫 Turbine AMQP，當時有個 spring-cloud-starter-turbine-amqp 依賴能夠用，裏邊包裝了相關的依賴，可是如今它被 deprecated 了，讓用 spring-cloud-starter-netflix-turbine-stream 來代替，這就得靠本身來組合了。而坑主要就出在這裏，至於哪些是必須的，哪些是添加了後就出問題的，還有依賴衝突的問題，都得靠本身來搞了。

JsonParseException

相關 issue：https://github.com/spring-cloud/spring-cloud-netflix/issues/2858

Turbine Stream 從 RabbitMQ 取數據的時候拋出如下異常：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59

com.fasterxml.jackson.core.JsonParseException: Unexpected character (',' (code 44)): Expected space separating root-level values at [Source: (String)"123,34,111,114,105,103,105,110,34,58,123,34,104,111,115,116,34,58,34,49,55,50,46,49,54,46,49,48,54,46,57,51,34,44,34,112,111,114,116,34,58,57,48,49,51,44,34,115,101,114,118,105,99,101,73,100,34,58,34,101,117,114,101,107,97,45,99,111,110,115,117,109,101,114,45,104,121,115,116,114,105,120,34,44,34,105,100,34,58,34,97,112,112,108,105,99,97,116,105,111,110,45,49,34,125,44,34,101,118,101,110,116,34,58,34,109,101,115,115,97,103,101,34,44,34,100,97,116,97,34,58,123,34,116,121,112,101,34,58,34,72,121,11"[truncated 6105 chars]; line: 1, column: 5] at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1798) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:663) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:561) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.base.ParserMinimalBase._reportMissingRootWS(ParserMinimalBase.java:608) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._verifyRootSpace(ReaderBasedJsonParser.java:1654) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._parsePosNumber(ReaderBasedJsonParser.java:1297) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:744) ~[jackson-core-2.9.3.jar:2.9.3] at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4129) ~[jackson-databind-2.9.3.jar:2.9.3] at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3988) ~[jackson-databind-2.9.3.jar:2.9.3] at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2992) ~[jackson-databind-2.9.3.jar:2.9.3] at org.springframework.cloud.netflix.turbine.stream.HystrixStreamAggregator.sendToSubject(HystrixStreamAggregator.java:73) ~[spring-cloud-netflix-turbine-stream-2.0.0.M8.jar:2.0.0.M8] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_161] at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_161] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_161] at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_161] at org.springframework.expression.spel.support.ReflectiveMethodExecutor.execute(ReflectiveMethodExecutor.java:120) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.ast.MethodReference.getValueInternal(MethodReference.java:111) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.ast.MethodReference.access$000(MethodReference.java:54) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:384) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:89) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.ast.SpelNodeImpl.getTypedValue(SpelNodeImpl.java:116) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:371) [spring-expression-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:169) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:157) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.util.MessagingMethodInvokerHelper.invokeExpression(MessagingMethodInvokerHelper.java:614) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.util.MessagingMethodInvokerHelper.invokeHandlerMethod(MessagingMethodInvokerHelper.java:605) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.util.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:468) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.util.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:312) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:104) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:93) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:158) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181) [spring-messaging-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160) [spring-messaging-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) [spring-messaging-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108) [spring-messaging-5.0.5.RELEASE.jar:5.0.5.RELEASE] at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203) [spring-integration-core-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1100(AmqpInboundChannelAdapter.java:59) [spring-integration-amqp-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:217) [spring-integration-amqp-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) ~[spring-retry-1.2.2.RELEASE.jar:na] at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) ~[spring-retry-1.2.2.RELEASE.jar:na] at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:214) [spring-integration-amqp-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1414) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1337) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1324) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1303) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:785) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:769) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1010) ~[spring-rabbit-2.0.3.RELEASE.jar:2.0.3.RELEASE] at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_161]

實際的 json 串相似如下格式：

1

{"origin":{"host":"172.16.106.93","port":9013,"serviceId":"eureka-consumer-hystrix","id":"application-1"},"event":"message","data":{"type":"HystrixCommand","name":"eureka-consumer-hystrix.HelloRemote#hello(String)","group":"eureka-producer","currentTime":1523938311789,"isCircuitBreakerOpen":false,"errorPercentage":0,"errorCount":0,"requestCount":0,"rollingCountCollapsedRequests":0,"rollingCountExceptionsThrown":0,"rollingCountFailure":0,"rollingCountFallbackFailure":0,"rollingCountFallbackRejection":0,"rollingCountFallbackSuccess":0,"rollingCountResponsesFromCache":0,"rollingCountSemaphoreRejected":0,"rollingCountShortCircuited":0,"rollingCountSuccess":0,"rollingCountThreadPoolRejected":0,"rollingCountTimeout":0,"currentConcurrentExecutionCount":1,"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":"THREAD","propertyValue_executionIsolationThreadTimeoutInMilliseconds":1000,"propertyValue_executionIsolationThreadInterruptOnTimeout":true,"propertyValue_executionIsolationThreadPoolKeyOverride":null,"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests":10,"propertyValue_metricsRollingStatisticalWindowInMilliseconds":10000,"propertyValue_requestCacheEnabled":true,"propertyValue_requestLogEnabled":true,"reportingHosts":1}}

上邊這串字符串轉爲 byte[] 後就是

1

123,34,111,114,105,103,105,110,34,58,123,34,104,111,115,116,34,58,34,49,55,50,46,49,54,46,49,48,54,46,57,51,34,44,34,112,111,114,116,34,58,57,48,49,51,44,34,115,101,114,118,105,99,101,73,100,34,58,34,101,117,114,101,107,97,45,99,111,110,115,117,109,101,114,45,104,121,115,116,114,105,120,34,44,34,105,100,34,58,34,97,112,112,108,105,99,97,116,105,111,110,45,49,34,125,44,34,101,118,101,110,116,34,58,34,109,101,115,115,97,103,101,34,44,34,100,97,116,97,34,58,123,34,116,121,112,101,34,58,34,72,121,115,116,114,105,120,67,111,109,109,97,110,100,34,44,34,110,97,109,101,34,58,34,101,117,114,101,107,97,45,99,111,110,115,117,109,101,114,45,104,121,115,116,114,105,120,46,72,101,108,108,111,82,101,109,111,116,101,35,104,101,108,108,111,40,83,116,114,105,110,103,41,34,44,34,103,114,111,117,112,34,58,34,101,117,114,101,107,97,45,112,114,111,100,117,99,101,114,34,44,34,99,117,114,114,101,110,116,84,105,109,101,34,58,49,53,50,51,57,51,56,56,57,49,55,53,48,44,34,105,115,67,105,114,99,117,105,116,66,114,101,97,107,101,114,79,112,101,110,34,58,102,97,108,115,101,44,34,101,114,114,111,114,80,101,114,99,101,110,116,97,103,101,34,58,48,44,34,101,114,114,111,114,67,111,117,110,116,34,58,48,44,34,114,101,113,117,101,115,116,67,111,117,110,116,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,67,111,108,108,97,112,115,101,100,82,101,113,117,101,115,116,115,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,69,120,99,101,112,116,105,111,110,115,84,104,114,111,119,110,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,70,97,105,108,117,114,101,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,70,97,108,108,98,97,99,107,70,97,105,108,117,114,101,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,70,97,108,108,98,97,99,107,82,101,106,101,99,116,105,111,110,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,70,97,108,108,98,97,99,107,83,117,99,99,101,115,115,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,82,101,115,112,111,110,115,101,115,70,114,111,109,67,97,99,104,101,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,83,101,109,97,112,104,111,114,101,82,101,106,101,99,116,101,100,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,83,104,111,114,116,67,105,114,99,117,105,116,101,100,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,83,117,99,99,101,115,115,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,84,104,114,101,97,100,80,111,111,108,82,101,106,101,99,116,101,100,34,58,48,44,34,114,111,108,108,105,110,103,67,111,117,110,116,84,105,109,101,111,117,116,34,58,48,44,34,99,117,114,114,101,110,116,67,111,110,99,117,114,114,101,110,116,69,120,101,99,117,116,105,111,110,67,111,117,110,116,34,58,49,44,34,108,97,116,101,110,99,121,69,120,101,99,117,116,101,95,109,101,97,110,34,58,48,44,34,108,97,116,101,110,99,121,69,120,101,99,117,116,101,34,58,123,34,48,34,58,48,44,34,50,53,34,58,48,44,34,53,48,34,58,48,44,34,55,53,34,58,48,44,34,57,48,34,58,48,44,34,57,53,34,58,48,44,34,57,57,34,58,48,44,34,57,57,46,53,34,58,48,44,34,49,48,48,34,58,48,125,44,34,108,97,116,101,110,99,121,84,111,116,97,108,95,109,101,97,110,34,58,48,44,34,108,97,116,101,110,99,121,84,111,116,97,108,34,58,123,34,48,34,58,48,44,34,50,53,34,58,48,44,34,53,48,34,58,48,44,34,55,53,34,58,48,44,34,57,48,34,58,48,44,34,57,53,34,58,48,44,34,57,57,34,58,48,44,34,57,57,46,53,34,58,48,44,34,49,48,48,34,58,48,125,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,82,101,113,117,101,115,116,86,111,108,117,109,101,84,104,114,101,115,104,111,108,100,34,58,50,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,83,108,101,101,112,87,105,110,100,111,119,73,110,77,105,108,108,105,115,101,99,111,110,100,115,34,58,53,48,48,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,69,114,114,111,114,84,104,114,101,115,104,111,108,100,80,101,114,99,101,110,116,97,103,101,34,58,53,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,70,111,114,99,101,79,112,101,110,34,58,102,97,108,115,101,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,70,111,114,99,101,67,108,111,115,101,100,34,58,102,97,108,115,101,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,99,105,114,99,117,105,116,66,114,101,97,107,101,114,69,110,97,98,108,101,100,34,58,116,114,117,101,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,101,120,101,99,117,116,105,111,110,73,115,111,108,97,116,105,111,110,83,116,114,97,116,101,103,121,34,58,34,84,72,82,69,65,68,34,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,101,120,101,99,117,116,105,111,110,73,115,111,108,97,116,105,111,110,84,104,114,101,97,100,84,105,109,101,111,117,116,73,110,77,105,108,108,105,115,101,99,111,110,100,115,34,58,49,48,48,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,101,120,101,99,117,116,105,111,110,73,115,111,108,97,116,105,111,110,84,104,114,101,97,100,73,110,116,101,114,114,117,112,116,79,110,84,105,109,101,111,117,116,34,58,116,114,117,101,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,101,120,101,99,117,116,105,111,110,73,115,111,108,97,116,105,111,110,84,104,114,101,97,100,80,111,111,108,75,101,121,79,118,101,114,114,105,100,101,34,58,110,117,108,108,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,101,120,101,99,117,116,105,111,110,73,115,111,108,97,116,105,111,110,83,101,109,97,112,104,111,114,101,77,97,120,67,111,110,99,117,114,114,101,110,116,82,101,113,117,101,115,116,115,34,58,49,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,102,97,108,108,98,97,99,107,73,115,111,108,97,116,105,111,110,83,101,109,97,112,104,111,114,101,77,97,120,67,111,110,99,117,114,114,101,110,116,82,101,113,117,101,115,116,115,34,58,49,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,109,101,116,114,105,99,115,82,111,108,108,105,110,103,83,116,97,116,105,115,116,105,99,97,108,87,105,110,100,111,119,73,110,77,105,108,108,105,115,101,99,111,110,100,115,34,58,49,48,48,48,48,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,114,101,113,117,101,115,116,67,97,99,104,101,69,110,97,98,108,101,100,34,58,116,114,117,101,44,34,112,114,111,112,101,114,116,121,86,97,108,117,101,95,114,101,113,117,101,115,116,76,111,103,69,110,97,98,108,101,100,34,58,116,114,117,101,44,34,114,101,112,111,114,116,105,110,103,72,111,115,116,115,34,58,49,125,125

從上邊異常能夠看出，這其實就是將這個 byte[] 轉爲 String 的時候出錯了。

在源碼裏找了很久，最後發現原來是啓動的時候要初始化一個 ConfigurableCompositeMessageConverter，可是這個類默認的只提供如下 4 個MessageConverter：

• MappingJackson2MessageConverter
• ByteArrayMessageConverter
• ObjectStringMessageConverter
• GenericMessageConverter

這 4 個 Converter 處理 byte[] -> String 的時候都會出問題（上邊的異常就是MappingJackson2MessageConverter這個拋出的）。說到這你可能會問了，byte[] 轉 String 有這麼難嗎？不就一個 new String(bytes) 就解決了。我也這麼想啊，就差本身動手寫了。

這時候發現了CompositeMessageConverterFactory，從名字上能夠看出就是 MessageConverter的工廠類，彷佛是一根救命稻草。它裏邊默認提供了 7 個 Converter，第一個就是ApplicationJsonMessageMarshallingConverter，看了裏邊的實現，這不正是我須要的嘛！

1 2 3 4 5 6

if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) { result = new String((byte[])message.getPayload(), StandardCharsets.UTF_8); } else { result = super.convertFromInternal(message, targetClass, conversionHint); }

因此就本身手動注入這個 bean

1 2 3 4

@Bean public ConfigurableCompositeMessageConverter integrationArgumentResolverMessageConverter(CompositeMessageConverterFactory factory) { return new ConfigurableCompositeMessageConverter(factory.getMessageConverterForAllRegistered().getConverters()); }

其實這個 Bean 在ContentTypeConfiguration中已經聲明瞭但沒效果，我只是原封不動的 copy 出來。

1 2 3 4

@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) public ConfigurableCompositeMessageConverter configurableCompositeMessageConverter(CompositeMessageConverterFactory factory){ return new ConfigurableCompositeMessageConverter(factory.getMessageConverterForAllRegistered().getConverters()); }

Fatal Exception thrown on Scheduler.Worker thread

這個的異常信息以下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

Exception in thread "RxComputationScheduler-1" java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.Worker thread. at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:59) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.AbstractMethodError at io.netty.util.ReferenceCountUtil.touch(ReferenceCountUtil.java:77) at io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:810) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:723) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:704) at io.netty.channel.DefaultChannelPipeline.write(DefaultChannelPipeline.java:1056) at io.netty.channel.AbstractChannel.write(AbstractChannel.java:290) at io.reactivex.netty.channel.DefaultChannelWriter.writeOnChannel(DefaultChannelWriter.java:165) at io.reactivex.netty.protocol.http.server.HttpServerResponse.writeOnChannel(HttpServerResponse.java:195) at io.reactivex.netty.channel.DefaultChannelWriter.write(DefaultChannelWriter.java:83) at io.reactivex.netty.channel.DefaultChannelWriter.writeAndFlush(DefaultChannelWriter.java:65) at org.springframework.cloud.netflix.turbine.stream.TurbineStreamConfiguration.lambda$null$6(TurbineStreamConfiguration.java:106) at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:69) at rx.observers.Subscribers$5.onNext(Subscribers.java:235) at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395) at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355) at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846) at rx.internal.operators.OnSubscribeRefCount$2.onNext(OnSubscribeRefCount.java:120) at rx.internal.operators.OperatorPublish$PublishSubscriber.dispatch(OperatorPublish.java:585) at rx.internal.operators.OperatorPublish$PublishSubscriber.onNext(OperatorPublish.java:283) at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77) at rx.internal.operators.OnSubscribeTimerPeriodically$1.call(OnSubscribeTimerPeriodically.java:52) at rx.internal.schedulers.SchedulePeriodicHelper$1.call(SchedulePeriodicHelper.java:72) at rx.internal.schedulers.EventLoopsScheduler$EventLoopWorker$2.call(EventLoopsScheduler.java:189) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) ... 7 more

同時在 Hystrix Dashboard 中 monitor 相應的地址的會提示 「Unable to connect to Command Metric Stream.」 並在 Console 裏報錯：

1	EventSource's response has a MIME type ("text/plain") that is not "text/event-stream". Aborting the connection.

這個問題真不知道怎麼搞了，先棄坑了，有空了再研究吧。

2018-05-06 更新：
這個問題能夠詳見 Github 上的 這個 issus 感謝 @MadeInChina

爲了本文的完整性，我這裏也說明下這個問題的解決思路和解決辦法。

這個異常信息關鍵的一行是是第 22 行

1	at org.springframework.cloud.netflix.turbine.stream.TurbineStreamConfiguration.lambda$null$6(TurbineStreamConfiguration.java:106)

咱們就打開 TurbineStreamConfiguration 看一下這 106 行到底寫了點啥

若是這時你點進去看 ServerSentEvent 這個類，問題其實就明瞭了（我當時就是困在了response.writeAndFlush這個方法上，根本沒注意到 ServerSentEvent）

這個類直接報錯，由於沒有完整實現 ByteBufHolder 裏邊的方法，從這你已經能看出來應該是相關依賴的問題了。咱們這裏注意一下這個類所在的 JAR 包 io.reactivex:rxnetty:0.4.9，而後再看一下它實現的這個接口

這個接口的變動記錄能夠看這裏。最後一次的修改是在 2016.5.17 添加了幾個方法，也就是說 ByteBufHolder 在io.netty:netty-buffer:4.1.0.Final（2016.5.25）就已經被修改了，更不用說io.netty:netty-buffer:4.1.23.Final（2018.4.4）了。而io.reactivex:rxnetty:0.4.9是一個 2015.5.6 發佈的 jar 包。

那咱們就用io.reactivex:rxnetty:0.4.20最新的版原本試試，pom.xml 裏添加如下內容

1 2 3 4 5 6

<dependency> <groupId>io.reactivex</groupId> <artifactId>rxnetty</artifactId> <version>0.4.20</version> <scope>runtime</scope> </dependency>

再次啓動並測試

1 2	$ curl http://localhost:8989 curl: (52) Empty reply from server

發現依舊報錯，控制檯裏錯誤以下

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76

2018-05-06 23:29:37.056 WARN 52477 --- [o-eventloop-3-2] i.n.c.AbstractChannelHandlerContext : An exception '{}' [enable DEBUG level for full stacktrace] was thrown by a user handler's exceptionCaught() method while handling the following exception: java.lang.NoSuchMethodError: rx.internal.operators.NotificationLite.instance()Lrx/internal/operators/NotificationLite; at io.reactivex.netty.protocol.http.UnicastContentSubject$State$BufferedObserver.<init>(UnicastContentSubject.java:243) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.UnicastContentSubject$State$BufferedObserver.<init>(UnicastContentSubject.java:241) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.UnicastContentSubject$State.<init>(UnicastContentSubject.java:197) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.UnicastContentSubject.create(UnicastContentSubject.java:132) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.UnicastContentSubject.create(UnicastContentSubject.java:122) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.UnicastContentSubject.create(UnicastContentSubject.java:117) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.server.ServerRequestResponseConverter$RequestState.createRxRequest(ServerRequestResponseConverter.java:176) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.server.ServerRequestResponseConverter$RequestState.access$100(ServerRequestResponseConverter.java:168) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.server.ServerRequestResponseConverter.channelRead(ServerRequestResponseConverter.java:87) ~[rx-netty-0.3.18.jar:na] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310) [netty-codec-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284) [netty-codec-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:241) [netty-handler-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.reactivex.netty.metrics.BytesInspector.channelRead(BytesInspector.java:59) [rx-netty-0.3.18.jar:na] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) [netty-common-4.1.23.Final.jar:4.1.23.Final] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.23.Final.jar:4.1.23.Final] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171] 2018-05-06 23:29:37.057 ERROR 52477 --- [o-eventloop-3-2] i.r.netty.server.DefaultErrorHandler : Unexpected error in RxNetty. java.lang.NullPointerException: null at io.reactivex.netty.protocol.http.server.ServerRequestResponseConverter.invokeContentOnNext(ServerRequestResponseConverter.java:160) ~[rx-netty-0.3.18.jar:na] at io.reactivex.netty.protocol.http.server.ServerRequestResponseConverter.channelRead(ServerRequestResponseConverter.java:96) ~[rx-netty-0.3.18.jar:na] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310) [netty-codec-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284) [netty-codec-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:241) [netty-handler-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.reactivex.netty.metrics.BytesInspector.channelRead(BytesInspector.java:59) [rx-netty-0.3.18.jar:na] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) [netty-transport-4.1.23.Final.jar:4.1.23.Final] at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) [netty-common-4.1.23.Final.jar:4.1.23.Final] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.23.Final.jar:4.1.23.Final] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]

就僅看異常棧的第一行，你會發現 UnicastContentSubject 這個類在當前環境下有三個

將三個都打開來看一下，會發現io.reactivex:rxnetty那兩個看起來問題不大，而com.netflix.rxnetty:rx-netty這個直接又是報錯，依舊是未徹底實現接口。

根據 DEBUG 的信息來看，而實際使用的就是com.netflix.rxnetty:rx-netty:0.3.18裏邊的這個，那就分析一下依賴吧，把這個殘疾的老古董（2014.11.6）給排除掉

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39

$ mvn dependency:tree -Dverbose -Dincludes=*:*netty* [INFO] Scanning for projects... [INFO] [INFO] ----------< com.windmt:spring-cloud-turbine-stream-rabbitmq >----------- [INFO] Building turbine-stream-rabbitmq 0.0.1-SNAPSHOT [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- maven-dependency-plugin:3.0.2:tree (default-cli) @ spring-cloud-turbine-stream-rabbitmq --- [INFO] Verbose not supported since maven-dependency-plugin 3.0 [INFO] com.windmt:spring-cloud-turbine-stream-rabbitmq:jar:0.0.1-SNAPSHOT [INFO] +- org.springframework.cloud:spring-cloud-starter-netflix-turbine-stream:jar:2.0.0.RC1:compile [INFO] | +- org.springframework.cloud:spring-cloud-starter-netflix-eureka-client:jar:2.0.0.RC1:compile [INFO] | | \- org.springframework.cloud:spring-cloud-starter-netflix-ribbon:jar:2.0.0.RC1:compile [INFO] | | \- com.netflix.ribbon:ribbon:jar:2.2.5:compile [INFO] | | +- com.netflix.ribbon:ribbon-transport:jar:2.2.5:runtime [INFO] | | | +- io.reactivex:rxnetty-contexts:jar:0.4.9:runtime [INFO] | | | \- io.reactivex:rxnetty-servo:jar:0.4.9:runtime [INFO] | | \- io.reactivex:rxnetty:jar:0.4.9:runtime [INFO] | \- com.netflix.turbine:turbine-core:jar:2.0.0-DP.2:compile [INFO] | \- com.netflix.rxnetty:rx-netty:jar:0.3.18:compile [INFO] | +- io.netty:netty-codec-http:jar:4.1.23.Final:compile [INFO] | | \- io.netty:netty-codec:jar:4.1.23.Final:compile [INFO] | \- io.netty:netty-transport-native-epoll:jar:4.1.23.Final:compile [INFO] | +- io.netty:netty-common:jar:4.1.23.Final:compile [INFO] | +- io.netty:netty-buffer:jar:4.1.23.Final:compile [INFO] | +- io.netty:netty-transport-native-unix-common:jar:4.1.23.Final:compile [INFO] | \- io.netty:netty-transport:jar:4.1.23.Final:compile [INFO] | \- io.netty:netty-resolver:jar:4.1.23.Final:compile [INFO] \- org.springframework.cloud:spring-cloud-stream-binder-rabbit:jar:2.0.0.RELEASE:compile [INFO] \- org.springframework.boot:spring-boot-starter-amqp:jar:2.0.1.RELEASE:compile [INFO] \- org.springframework.amqp:spring-rabbit:jar:2.0.3.RELEASE:compile [INFO] \- com.rabbitmq:http-client:jar:2.0.1.RELEASE:compile [INFO] \- io.projectreactor.ipc:reactor-netty:jar:0.7.6.RELEASE:compile [INFO] +- io.netty:netty-handler:jar:4.1.23.Final:compile [INFO] \- io.netty:netty-handler-proxy:jar:4.1.23.Final:compile [INFO] \- io.netty:netty-codec-socks:jar:4.1.23.Final:compile [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------

最終 pom.xml 裏的依賴座標以下（這也是能正常啓動 Turbine Stream 的最小配置了，Spring Cloud 的版本爲 Finchley.RC1）：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-turbine-stream</artifactId> <exclusions> <exclusion> <groupId>com.netflix.rxnetty</groupId> <artifactId>rx-netty</artifactId> </exclusion> <exclusion> <groupId>io.reactivex</groupId> <artifactId>rxnetty</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>io.reactivex</groupId> <artifactId>rxnetty</artifactId> <version>0.4.20</version> <scope>runtime</scope> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>

再次測試就能正常收到 SSE 了

1 2 3 4 5 6

$ curl http://localhost:8989 event: message data: {"rollingCountFallbackFailure":0,"rollingCountFallbackSuccess":0,"propertyValue_circuitBreakerRequestVolumeThreshold":"20","propertyValue_circuitBreakerForceOpen":false,"propertyValue_metricsRollingStatisticalWindowInMilliseconds":"10000","latencyTotal_mean":0,"type":"HystrixCommand","rollingCountResponsesFromCache":0,"TypeAndName":"TypeAndName=>HystrixCommand_consumer-feign-hystrix-stream.HelloRemote#hello(String)","rollingCountTimeout":0,"propertyValue_executionIsolationStrategy":"THREAD","instanceId":"application-1","rollingCountFailure":0,"rollingCountExceptionsThrown":0,"latencyExecute_mean":0,"isCircuitBreakerOpen":false,"errorCount":0,"group":"producer","rollingCountSemaphoreRejected":0,"latencyTotal":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"requestCount":0,"rollingCountCollapsedRequests":0,"rollingCountShortCircuited":0,"latencyExecute":{"0":0,"25":0,"50":0,"75":0,"90":0,"95":0,"99":0,"99.5":0,"100":0},"propertyValue_circuitBreakerSleepWindowInMilliseconds":"5000","currentConcurrentExecutionCount":0,"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests":"10","errorPercentage":0,"rollingCountThreadPoolRejected":0,"propertyValue_circuitBreakerEnabled":true,"propertyValue_executionIsolationThreadInterruptOnTimeout":true,"propertyValue_requestCacheEnabled":true,"rollingCountFallbackRejection":0,"propertyValue_requestLogEnabled":true,"rollingCountSuccess":0,"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests":"10","InstanceKey":"InstanceKey=>application-1","propertyValue_circuitBreakerErrorThresholdPercentage":"50","propertyValue_circuitBreakerForceClosed":false,"name":"consumer-feign-hystrix-stream.HelloRemote#hello(String)","reportingHosts":1,"propertyValue_executionIsolationThreadPoolKeyOverride":"null","propertyValue_executionIsolationThreadTimeoutInMilliseconds":"1000"} event: message data: {"currentCorePoolSize":10,"currentLargestPoolSize":8,"currentActiveCount":0,"propertyValue_metricsRollingStatisticalWindowInMilliseconds":"10000","currentMaximumPoolSize":10,"currentQueueSize":0,"type":"HystrixThreadPool","currentTaskCount":8,"TypeAndName":"TypeAndName=>HystrixThreadPool_producer","currentCompletedTaskCount":8,"rollingMaxActiveThreads":0,"instanceId":"application-1","InstanceKey":"InstanceKey=>application-1","name":"producer","reportingHosts":1,"currentPoolSize":8,"propertyValue_queueSizeRejectionThreshold":"5","rollingCountThreadsExecuted":0}

控制檯也不報錯了

1 2 3 4

2018-05-06 23:57:25.833 INFO 53309 --- [o-eventloop-3-1] o.s.c.n.t.s.TurbineStreamConfiguration : SSE Request Received 2018-05-06 23:57:25.843 INFO 53309 --- [o-eventloop-3-1] o.s.c.n.t.s.TurbineStreamConfiguration : Starting aggregation 2018-05-06 23:57:32.115 INFO 53309 --- [o-eventloop-3-1] o.s.c.n.t.s.TurbineStreamConfiguration : Unsubscribing RxNetty server connection 2018-05-06 23:57:32.116 INFO 53309 --- [o-eventloop-3-1] o.s.c.n.t.s.TurbineStreamConfiguration : Unsubscribing aggregation.

而後在 Hystrix Dashboard 的地址欄裏輸入 http://localhost:8989 就能看到了（/turbine.stream可加可不加，若是要修改端口號，在 Spring Boot 的配置文件中修改 turbine.stream.port）