Merge branch 'master' into patch-2

pom.xml

@@ -100,7 +100,6 @@
         <module>spring-cloud-alicloud-oss</module>
         <module>spring-cloud-alicloud-acm</module>
         <module>spring-cloud-alicloud-ans</module>
-        <module>spring-cloud-starter-bus-rocketmq</module>
     </modules>
 
     <dependencyManagement>

spring-cloud-alibaba-docs/src/main/asciidoc-zh/nacos-config.adoc

@@ -12,7 +12,7 @@ Spring Cloud Alibaba Nacos Config 是 Config Server 和 Client 的替代方案
 
 1、启动Nacos Server。启动方式可见 https://nacos.io/zh-cn/docs/quick-start.html[Nacos 官网]
 
-2、启动好Nacos之后，在Nacos添加如下的配置。注意dataid是以 properties为扩展名，默认的文件扩展名方式。如下所示：
+2、启动好Nacos之后，在Nacos添加如下的配置：
 
 [source,subs="normal"]
 ----
@@ -20,15 +20,17 @@ Data ID:    nacos-config.properties
 
 Group  :    DEFAULT_GROUP
 
-配置格式:    TEXT
+配置格式:    Properties
 
-配置内容：   user.name: nacos-config-properties
-            user.age: 90
+配置内容：   user.name=nacos-config-properties
+            user.age=90
 ----
 
+NOTE: 注意dataid是以 properties(默认的文件扩展名方式)为扩展名。
+
 ==== 客户端使用方式
 
-为了能够在应用程序使用Nacos实现应用的外部化配置，在构建应用的同时添加一个Spring Boot Starter org.springframework.cloud:spring-cloud-starter-alibaba-nacos-config。以下是一个基础的maven 依赖配置:
+为了能够在应用程序中使用 Nacos 来实现应用的外部化配置，在构建应用的同时添加一个Spring Boot Starter org.springframework.cloud:spring-cloud-starter-alibaba-nacos-config。以下是一个基本的 maven 依赖配置:
 
 [source,xml]
 ----
@@ -95,7 +97,7 @@ public class ProviderApplication {
 }
 ----
 
-spring-cloud-starter-alibaba-nacos-config 对于 Nacos 服务端的基础配置没有默认值，因此在运行此Example 之前， 必须使用 bootstrap.properties 配置文件来配置Nacos Server地址，例如：
+在运行此 Example 之前， 必须使用 bootstrap.properties 配置文件来配置Nacos Server 地址，例如：
 
 .bootstrap.properties
 [source,properties]
@@ -108,7 +110,7 @@ NOTE: 注意当你使用域名的方式来访问 Nacos 时，`spring.cloud.nacos
 例如 Nacos 的域名为abc.com.nacos，监听的端口为 80，则 `spring.cloud.nacos.config.server-addr=abc.com.nacos:80`。
 注意 80 端口不能省略。
 
-启动这个Example，可以在控制台看到打印出的值正是在Nacos上预先配置好的值。
+启动这个 Example，可以看到如下输出结果：
 
 [source,subs="normal"]
 ----
@@ -120,7 +122,7 @@ user name :nacos-config-properties; age: 90
 
 === 基于 dataid 为 yaml 的文件扩展名配置方式
 
-spring-cloud-starter-alibaba-nacos-config 对于 yaml 格式也是完美支持。这个时候只需要完成以下两步：
+spring-cloud-starter-alibaba-nacos-config 对于 yaml 格式也是完美支持的。这个时候只需要完成以下两步：
 
 1、在应用的 bootstrap.properties 配置文件中显示的声明 dataid 文件扩展名。如下所示
 
@@ -144,7 +146,7 @@ Group  :        DEFAULT_GROUP
                 user.age: 68
 ----
 
-这两步完成后，重启测试程序，可以在控制台看到输出是以dataid为 nacos-config.yaml 配置的值。
+这两步完成后，重启测试程序，可以看到如下输出结果。
 
 [source,subs="normal"]
 ----
@@ -165,10 +167,9 @@ public class ProviderApplication {
     public static void main(String[] args) {
         ConfigurableApplicationContext applicationContext = SpringApplication.run(ProviderApplication.class, args);
         while(true) {
+            //当动态配置刷新时，会更新到 Enviroment中，因此这里每隔一秒中从Enviroment中获取配置
             String userName = applicationContext.getEnvironment().getProperty("user.name");
             String userAge = applicationContext.getEnvironment().getProperty("user.age");
-            //获取当前部署的环境
-            String currentEnv = applicationContext.getEnvironment().getProperty("current.env");
             System.err.println("user name :" + userName + "; age: " + userAge);
             TimeUnit.SECONDS.sleep(1);
         }
@@ -195,13 +196,15 @@ NOTE: 你可以通过配置 `spring.cloud.nacos.config.refresh.enabled=false`
 
 === 可支持profile粒度的配置
 
-spring-cloud-starter-alibaba-nacos-config 在加载配置的时候，不仅仅加载了以 dataid 为 `${spring.application.name}.${file-extension:properties}`   为前缀的基础配置，还加载了dataid为 `${spring.application.name}-${profile}.${file-extension:properties}` 的基础配置。在日常开发中如果遇到多套环境下的不同配置，可以打开 Spring 自带的配置功能。
+spring-cloud-starter-alibaba-nacos-config 在加载配置的时候，不仅仅加载了以 dataid 为 `${spring.application.name}.${file-extension:properties}`   为前缀的基础配置，还加载了dataid为 `${spring.application.name}-${profile}.${file-extension:properties}` 的基础配置。在日常开发中如果遇到多套环境下的不同配置，可以通过Spring 提供的 `${spring.profiles.active}` 这个配置项来配置。
 
 [source,properties]
 ----
 spring.profiles.active=develop
 ----
 
+NOTE: ${spring.profiles.active} 当通过配置文件来指定时必须放在 bootstrap.properties 文件中。
+
 Nacos 上新增一个dataid为：nacos-config-develop.yaml的基础配置，如下所示：
 
 [source,subs="normal"]
@@ -250,7 +253,7 @@ in develop-evn enviroment; user name :nacos-config-yaml-update; age: 68
 spring.profiles.active=product
 ----
 
-同时生产环境上Nacos需要添加该对应 dataid 的基础配置。例如，在生成环境下的 Naocs 添加了dataid为：nacos-config-product.yaml的配置：
+同时生产环境上 Nacos 需要添加对应 dataid 的基础配置。例如，在生成环境下的 Naocs 添加了dataid为：nacos-config-product.yaml的配置：
 
 [source,subs="normal"]
 ----
@@ -271,6 +274,7 @@ in product-env enviroment; user name :nacos-config-yaml-update; age: 68
 2018-11-02 15:42:14.628  INFO 33024 --- [Thread-11] ConfigServletWebServerApplicationContext : Closing org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@6aa8e115: startup date [Fri Nov 02 15:42:03 CST 2018]; parent: org.springframework.context.annotation.AnnotationConfigApplicationContext@19bb07ed
 ----
 
+
 NOTE: 此案例中我们通过 `spring.profiles.active=****` 的方式写死在配置文件中，而在真正的项目实施过程中这个变量的值是需要不同环境而有不同的值。这个时候通常的做法是通过 `-Dspring.profiles.active=****` 参数指定其配置来达到环境间灵活的切换。
 
 === 支持自定义 namespace 的配置

spring-cloud-alibaba-examples/rocketmq-example/pom.xml

@@ -31,6 +31,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md

@@ -122,12 +122,12 @@ spring.cloud.stream.bindings.output.destination=test-topic
 spring.cloud.stream.bindings.output.content-type=application/json
 
 spring.cloud.stream.bindings.input1.destination=test-topic
-spring.cloud.stream.bindings.input1.content-type=application/json
+spring.cloud.stream.bindings.input1.content-type=text/plain
 spring.cloud.stream.bindings.input1.group=test-group1
 spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true
 
 spring.cloud.stream.bindings.input2.destination=test-topic
-spring.cloud.stream.bindings.input2.content-type=application/json
+spring.cloud.stream.bindings.input2.content-type=text/plain
 spring.cloud.stream.bindings.input2.group=test-group2
 spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false
 spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr
@@ -195,7 +195,7 @@ Spring Boot 应用支持通过 Endpoint 来暴露相关信息，RocketMQ Stream
 * Spring Boot 1.x 中添加配置 `management.security.enabled=false`
 * Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`
 
-Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq-binder 来查看 RocketMQ Binder Endpoint 的信息。Spring Boot 2.x 可以通过访问 http://127.0.0.1:28081/acutator/rocketmq-binder 来访问。
+Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq_binder 来查看 RocketMQ Binder Endpoint 的信息。Spring Boot 2.x 可以通过访问 http://127.0.0.1:28081/actuator/rocketmq-binder 来访问。
 
 这里会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。
 
@@ -249,6 +249,14 @@ Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq-binder 来查
 }
 ```
 
+注意：要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core) 。如若不加，endpoint将会显示warning信息而不会显示统计信息：
+
+```json
+{
+    "warning": "please add metrics-core dependency, we use it to metrics"
+}
+```
+
 ## More
 
 RocketMQ 是一款功能强大的分布式消息系统，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java

@@ -6,14 +6,14 @@ package org.springframework.cloud.alibaba.cloud.examples;
 public class Foo {
 
 	private int id;
-	private String tag;
+	private String bar;
 
 	public Foo() {
 	}
 
-	public Foo(int id, String tag) {
+	public Foo(int id, String bar) {
 		this.id = id;
-		this.tag = tag;
+		this.bar = bar;
 	}
 
 	public int getId() {
@@ -24,16 +24,16 @@ public class Foo {
 		this.id = id;
 	}
 
-	public String getTag() {
-		return tag;
-	}
+    public String getBar() {
+        return bar;
+    }
 
-	public void setTag(String tag) {
-		this.tag = tag;
-	}
+    public void setBar(String bar) {
+        this.bar = bar;
+    }
 
-	@Override
+    @Override
 	public String toString() {
-		return "Foo{" + "id=" + id + ", tag='" + tag + '\'' + '}';
+		return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}';
 	}
 }

spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties

@@ -6,13 +6,13 @@ spring.cloud.stream.bindings.output.destination=test-topic
 spring.cloud.stream.bindings.output.content-type=application/json
 
 spring.cloud.stream.bindings.input1.destination=test-topic
-spring.cloud.stream.bindings.input1.content-type=application/json
+spring.cloud.stream.bindings.input1.content-type=text/plain
 spring.cloud.stream.bindings.input1.group=test-group1
 spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true
 spring.cloud.stream.bindings.input1.consumer.maxAttempts=1
 
 spring.cloud.stream.bindings.input2.destination=test-topic
-spring.cloud.stream.bindings.input2.content-type=application/json
+spring.cloud.stream.bindings.input2.content-type=text/plain
 spring.cloud.stream.bindings.input2.group=test-group2
 spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false
 spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr

spring-cloud-stream-binder-rocketmq/pom.xml

@@ -22,13 +22,16 @@
         </dependency>
 
         <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-core</artifactId>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-client</artifactId>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.0.3</version>
+            <scope>provided</scope>
+            <optional>true</optional>
         </dependency>
 
         <dependency>

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java

@@ -5,6 +5,8 @@ package org.springframework.cloud.stream.binder.rocketmq;
  */
 public interface RocketMQBinderConstants {
 
+	String ENDPOINT_ID = "rocketmq-binder";
+
 	/**
 	 * Header key
 	 */
@@ -17,10 +19,27 @@ public interface RocketMQBinderConstants {
 	String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
 
 	/**
-	 * Instrumentation key
+	 * Instrumentation
 	 */
 	String LASTSEND_TIMESTAMP = "lastSend.timestamp";
 
-	String ENDPOINT_ID = "rocketmq-binder";
+	interface Metrics {
+		interface Producer {
+			String PREFIX = "scs-rocketmq.producer.";
+			String TOTAL_SENT = "totalSent";
+			String TOTAL_SENT_FAILURES = "totalSentFailures";
+			String SENT_PER_SECOND = "sentPerSecond";
+			String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
+		}
+
+		interface Consumer {
+			String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
+			String PREFIX = "scs-rocketmq.consumer.";
+			String TOTAL_CONSUMED = "totalConsumed";
+			String CONSUMED_PER_SECOND = "consumedPerSecond";
+			String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
+			String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
+		}
+	}
 
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java

@@ -34,7 +34,6 @@ public class RocketMQMessageChannelBinder extends
 			.getLogger(RocketMQMessageChannelBinder.class);
 
 	private final RocketMQExtendedBindingProperties extendedBindingProperties;
-	private final RocketMQTopicProvisioner rocketTopicProvisioner;
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 	private final InstrumentationManager instrumentationManager;
 	private final ConsumersManager consumersManager;
@@ -47,7 +46,6 @@ public class RocketMQMessageChannelBinder extends
 		super(null, provisioningProvider);
 		this.consumersManager = consumersManager;
 		this.extendedBindingProperties = extendedBindingProperties;
-		this.rocketTopicProvisioner = provisioningProvider;
 		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
 		this.instrumentationManager = instrumentationManager;
 	}
@@ -63,7 +61,7 @@ public class RocketMQMessageChannelBinder extends
 		}
 		else {
 			throw new RuntimeException("Binding for channel " + destination.getName()
-					+ "has been disabled, message can't be delivered");
+					+ " has been disabled, message can't be delivered");
 		}
 	}
 
@@ -74,7 +72,7 @@ public class RocketMQMessageChannelBinder extends
 			throws Exception {
 		if (group == null || "".equals(group)) {
 			throw new RuntimeException(
-					"'group' must be configured for channel + " + destination.getName());
+					"'group must be configured for channel + " + destination.getName());
 		}
 
 		RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java

@@ -4,12 +4,11 @@ import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderCon
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
 import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
-
-import com.codahale.metrics.MetricRegistry;
+import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
 
 /**
  * @author Timur Valiev
@@ -18,23 +17,22 @@ import com.codahale.metrics.MetricRegistry;
 @Endpoint(id = ENDPOINT_ID)
 public class RocketMQBinderEndpoint {
 
-	private MetricRegistry metricRegistry = new MetricRegistry();
-	private Map<String, Object> runtime = new ConcurrentHashMap<>();
+	@Autowired(required = false)
+	private InstrumentationManager instrumentationManager;
 
 	@ReadOperation
 	public Map<String, Object> invoke() {
 		Map<String, Object> result = new HashMap<>();
-		result.put("metrics", metricRegistry().getMetrics());
-		result.put("runtime", runtime());
+		if (instrumentationManager != null) {
+			result.put("metrics",
+					instrumentationManager.getMetricRegistry().getMetrics());
+			result.put("runtime", instrumentationManager.getRuntime());
+		}
+		else {
+			result.put("warning",
+					"please add metrics-core dependency, we use it for metrics");
+		}
 		return result;
 	}
 
-	public MetricRegistry metricRegistry() {
-		return metricRegistry;
-	}
-
-	public Map<String, Object> runtime() {
-		return runtime;
-	}
-
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java

@@ -1,5 +1,6 @@
 package org.springframework.cloud.stream.binder.rocketmq.actuator;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.actuate.health.AbstractHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
@@ -11,28 +12,33 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
  */
 public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
 
-	private final InstrumentationManager instrumentationManager;
-
-	public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
-		this.instrumentationManager = instrumentationManager;
-	}
+	@Autowired(required = false)
+	private InstrumentationManager instrumentationManager;
 
 	@Override
 	protected void doHealthCheck(Health.Builder builder) throws Exception {
-		if (instrumentationManager.getHealthInstrumentations().stream()
-				.allMatch(Instrumentation::isUp)) {
-			builder.up();
-			return;
+		if (instrumentationManager != null) {
+			if (instrumentationManager.getHealthInstrumentations().stream()
+					.allMatch(Instrumentation::isUp)) {
+				builder.up();
+				return;
+			}
+			if (instrumentationManager.getHealthInstrumentations().stream()
+					.allMatch(Instrumentation::isOutOfService)) {
+				builder.outOfService();
+				return;
+			}
+			builder.down();
+			instrumentationManager.getHealthInstrumentations().stream()
+					.filter(instrumentation -> !instrumentation.isStarted())
+					.forEach(instrumentation1 -> builder
+							.withException(instrumentation1.getStartException()));
 		}
-		if (instrumentationManager.getHealthInstrumentations().stream()
-				.allMatch(Instrumentation::isOutOfService)) {
-			builder.outOfService();
-			return;
+		else {
+			builder.down();
+			builder.withDetail("warning",
+					"please add metrics-core dependency, we use it for metrics");
 		}
-		builder.down();
-		instrumentationManager.getHealthInstrumentations().stream()
-				.filter(instrumentation -> !instrumentation.isStarted())
-				.forEach(instrumentation1 -> builder
-						.withException(instrumentation1.getStartException()));
+
 	}
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java

@@ -25,6 +25,9 @@ public class RocketMQBinderAutoConfiguration {
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
+	@Autowired(required = false)
+	private InstrumentationManager instrumentationManager;
+
 	@Autowired
 	public RocketMQBinderAutoConfiguration(
 			RocketMQExtendedBindingProperties extendedBindingProperties,
@@ -43,7 +46,6 @@ public class RocketMQBinderAutoConfiguration {
 	@Bean
 	public RocketMQMessageChannelBinder rocketMessageChannelBinder(
 			RocketMQTopicProvisioner provisioningProvider,
-			InstrumentationManager instrumentationManager,
 			ConsumersManager consumersManager) {
 		RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
 				consumersManager, extendedBindingProperties, provisioningProvider,
@@ -52,8 +54,7 @@ public class RocketMQBinderAutoConfiguration {
 	}
 
 	@Bean
-	public ConsumersManager consumersManager(
-			InstrumentationManager instrumentationManager) {
+	public ConsumersManager consumersManager() {
 		return new ConsumersManager(instrumentationManager,
 				rocketBinderConfigurationProperties);
 	}

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java

@@ -1,7 +1,9 @@
 package org.springframework.cloud.stream.binder.rocketmq.config;
 
 import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration;
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderEndpoint;
 import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@@ -13,6 +15,7 @@ import org.springframework.context.annotation.Configuration;
  */
 @Configuration
 @AutoConfigureAfter(EndpointAutoConfiguration.class)
+@ConditionalOnClass(Endpoint.class)
 public class RocketMQBinderEndpointAutoConfiguration {
 
 	@Bean
@@ -21,16 +24,14 @@ public class RocketMQBinderEndpointAutoConfiguration {
 	}
 
 	@Bean
-	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(
-			InstrumentationManager instrumentationManager) {
-		return new RocketMQBinderHealthIndicator(instrumentationManager);
+	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
+		return new RocketMQBinderHealthIndicator();
 	}
 
 	@Bean
-	public InstrumentationManager instrumentationManager(
-			RocketMQBinderEndpoint rocketBinderEndpoint) {
-		return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(),
-				rocketBinderEndpoint.runtime());
+	@ConditionalOnClass(name = "com.codahale.metrics.Counter")
+	public InstrumentationManager instrumentationManager() {
+		return new InstrumentationManager();
 	}
 
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java

@@ -3,6 +3,7 @@ package org.springframework.cloud.stream.binder.rocketmq.consuming;
 import java.util.AbstractMap;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -27,9 +28,10 @@ public class ConsumersManager {
 	private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
 	private final Map<String, Boolean> started = new HashMap<>();
 	private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap = new HashMap<>();
-	private final InstrumentationManager instrumentationManager;
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
+	private InstrumentationManager instrumentationManager;
+
 	public ConsumersManager(InstrumentationManager instrumentationManager,
 			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
 		this.instrumentationManager = instrumentationManager;
@@ -41,9 +43,12 @@ public class ConsumersManager {
 			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
 		propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic),
 				consumerProperties);
-		ConsumerGroupInstrumentation instrumentation = instrumentationManager
-				.getConsumerGroupInstrumentation(group);
-		instrumentationManager.addHealthInstrumentation(instrumentation);
+
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
+			ConsumerGroupInstrumentation instrumentation = manager
+					.getConsumerGroupInstrumentation(group);
+			instrumentationManager.addHealthInstrumentation(instrumentation);
+		});
 
 		if (consumerGroups.containsKey(group)) {
 			return consumerGroups.get(group);
@@ -87,16 +92,23 @@ public class ConsumersManager {
 		if (started.get(group)) {
 			return;
 		}
-		ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager
-				.getConsumerGroupInstrumentation(group);
-		instrumentationManager.addHealthInstrumentation(groupInstrumentation);
+
+		ConsumerGroupInstrumentation groupInstrumentation = null;
+		if (Optional.ofNullable(instrumentationManager).isPresent()) {
+			groupInstrumentation = instrumentationManager
+					.getConsumerGroupInstrumentation(group);
+			instrumentationManager.addHealthInstrumentation(groupInstrumentation);
+		}
+
 		try {
 			consumerGroups.get(group).start();
 			started.put(group, true);
-			groupInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(groupInstrumentation)
+					.ifPresent(g -> g.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			groupInstrumentation.markStartFailed(e);
+			Optional.ofNullable(groupInstrumentation)
+					.ifPresent(g -> g.markStartFailed(e));
 			logger.error("RocketMQ Consumer hasn't been started. Caused by "
 					+ e.getErrorMessage(), e);
 			throw e;

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java

@@ -4,6 +4,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -48,20 +49,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	private ConsumerInstrumentation consumerInstrumentation;
 
+	private InstrumentationManager instrumentationManager;
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<? extends Object> recoveryCallback;
+
 	private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
 
 	private final String destination;
 
 	private final String group;
 
-	private final InstrumentationManager instrumentationManager;
-
 	private final ConsumersManager consumersManager;
 
-	private RetryTemplate retryTemplate;
-
-	private RecoveryCallback<? extends Object> recoveryCallback;
-
 	public RocketMQInboundChannelAdapter(ConsumersManager consumersManager,
 			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
 			String destination, String group,
@@ -75,21 +76,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	@Override
 	protected void doStart() {
-		if (!consumerProperties.getExtension().getEnabled()) {
+		if (consumerProperties == null
+				|| !consumerProperties.getExtension().getEnabled()) {
 			return;
 		}
 
-		String tags = consumerProperties == null ? null
-				: consumerProperties.getExtension().getTags();
-		Boolean isOrderly = consumerProperties == null ? false
-				: consumerProperties.getExtension().getOrderly();
+		String tags = consumerProperties.getExtension().getTags();
+		Boolean isOrderly = consumerProperties.getExtension().getOrderly();
 
 		DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
 				destination, consumerProperties);
 
 		final CloudStreamMessageListener listener = isOrderly
-				? new CloudStreamMessageListenerOrderly(instrumentationManager)
-				: new CloudStreamMessageListenerConcurrently(instrumentationManager);
+				? new CloudStreamMessageListenerOrderly()
+				: new CloudStreamMessageListenerConcurrently();
 
 		if (retryTemplate != null) {
 			retryTemplate.registerListener(listener);
@@ -99,9 +99,10 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 				: Arrays.stream(tags.split("\\|\\|")).map(String::trim)
 						.collect(Collectors.toSet());
 
-		consumerInstrumentation = instrumentationManager
-				.getConsumerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
+			consumerInstrumentation = manager.getConsumerInstrumentation(destination);
+			manager.addHealthInstrumentation(consumerInstrumentation);
+		});
 
 		try {
 			if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) {
@@ -111,10 +112,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 			else {
 				consumer.subscribe(destination, String.join(" || ", tagsSet));
 			}
-			consumerInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(consumerInstrumentation)
+					.ifPresent(c -> c.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			consumerInstrumentation.markStartFailed(e);
+			Optional.ofNullable(consumerInstrumentation)
+					.ifPresent(c -> c.markStartFailed(e));
 			logger.error("RocketMQ Consumer hasn't been subscribed. Caused by "
 					+ e.getErrorMessage(), e);
 			throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e);
@@ -148,12 +151,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	protected class CloudStreamMessageListener implements MessageListener, RetryListener {
 
-		private final InstrumentationManager instrumentationManager;
-
-		CloudStreamMessageListener(InstrumentationManager instrumentationManager) {
-			this.instrumentationManager = instrumentationManager;
-		}
-
 		Acknowledgement consumeMessage(final List<MessageExt> msgs) {
 			boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
 			try {
@@ -180,23 +177,29 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 				}
 				else {
 					Acknowledgement result = doSendMsgs(msgs, null);
-					instrumentationManager
-							.getConsumerInstrumentation(
-									RocketMQInboundChannelAdapter.this.destination)
-							.markConsumed();
+					Optional.ofNullable(
+							RocketMQInboundChannelAdapter.this.instrumentationManager)
+							.ifPresent(manager -> {
+								manager.getConsumerInstrumentation(
+										RocketMQInboundChannelAdapter.this.destination)
+										.markConsumed();
+							});
 					return result;
 				}
 			}
 			catch (Exception e) {
 				logger.error(
-						"Rocket Message hasn't been processed successfully. Caused by ",
+						"RocketMQ Message hasn't been processed successfully. Caused by ",
 						e);
-				instrumentationManager
-						.getConsumerInstrumentation(
-								RocketMQInboundChannelAdapter.this.destination)
-						.markConsumedFailure();
+				Optional.ofNullable(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
+						.ifPresent(manager -> {
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumedFailure();
+						});
 				throw new RuntimeException(
-						"Rocket Message hasn't been processed successfully. Caused by ",
+						"RocketMQ Message hasn't been processed successfully. Caused by ",
 						e);
 			}
 		}
@@ -232,16 +235,22 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 		public <T, E extends Throwable> void close(RetryContext context,
 				RetryCallback<T, E> callback, Throwable throwable) {
 			if (throwable != null) {
-				instrumentationManager
-						.getConsumerInstrumentation(
-								RocketMQInboundChannelAdapter.this.destination)
-						.markConsumedFailure();
+				Optional.ofNullable(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
+						.ifPresent(manager -> {
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumedFailure();
+						});
 			}
 			else {
-				instrumentationManager
-						.getConsumerInstrumentation(
-								RocketMQInboundChannelAdapter.this.destination)
-						.markConsumed();
+				Optional.ofNullable(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
+						.ifPresent(manager -> {
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumed();
+						});
 			}
 		}
 
@@ -254,11 +263,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 	protected class CloudStreamMessageListenerConcurrently
 			extends CloudStreamMessageListener implements MessageListenerConcurrently {
 
-		public CloudStreamMessageListenerConcurrently(
-				InstrumentationManager instrumentationManager) {
-			super(instrumentationManager);
-		}
-
 		@Override
 		public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
 				ConsumeConcurrentlyContext context) {
@@ -272,11 +276,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 	protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener
 			implements MessageListenerOrderly {
 
-		public CloudStreamMessageListenerOrderly(
-				InstrumentationManager instrumentationManager) {
-			super(instrumentationManager);
-		}
-
 		@Override
 		public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
 				ConsumeOrderlyContext context) {

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java

@@ -2,6 +2,7 @@ package org.springframework.cloud.stream.binder.rocketmq.integration;
 
 import java.time.Instant;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -30,14 +31,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	private ProducerInstrumentation producerInstrumentation;
 
+	private InstrumentationManager instrumentationManager;
+
 	private final RocketMQProducerProperties producerProperties;
 
 	private final String destination;
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
-	private final InstrumentationManager instrumentationManager;
-
 	protected volatile boolean running = false;
 
 	public RocketMQMessageHandler(String destination,
@@ -54,9 +55,10 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 	public void start() {
 		producer = new DefaultMQProducer(destination);
 
-		producerInstrumentation = instrumentationManager
-				.getProducerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(producerInstrumentation);
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
+			producerInstrumentation = manager.getProducerInstrumentation(destination);
+			manager.addHealthInstrumentation(producerInstrumentation);
+		});
 
 		producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
 
@@ -66,10 +68,12 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 		try {
 			producer.start();
-			producerInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			producerInstrumentation.markStartFailed(e);
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markStartFailed(e));
 			logger.error(
 					"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
 			throw new MessagingException(e.getMessage(), e);
@@ -127,14 +131,16 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 				RocketMQMessageHeaderAccessor.putSendResult((MutableMessage) message,
 						sendRes);
 			}
-			instrumentationManager.getRuntime().put(
-					RocketMQBinderConstants.LASTSEND_TIMESTAMP,
-					Instant.now().toEpochMilli());
-			producerInstrumentation.markSent();
+			Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
+				manager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP,
+						Instant.now().toEpochMilli());
+			});
+			Optional.ofNullable(producerInstrumentation).ifPresent(p -> p.markSent());
 		}
 		catch (MQClientException | RemotingException | MQBrokerException
 				| InterruptedException | UnsupportedOperationException e) {
-			producerInstrumentation.markSentFailure();
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markSentFailure());
 			logger.error(
 					"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
 			throw new MessagingException(e.getMessage(), e);

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java

@@ -1,7 +1,5 @@
 package org.springframework.cloud.stream.binder.rocketmq.metrics;
 
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import com.codahale.metrics.MetricRegistry;
 
 /**
@@ -11,24 +9,9 @@ import com.codahale.metrics.MetricRegistry;
 public class ConsumerGroupInstrumentation extends Instrumentation {
 	private MetricRegistry metricRegistry;
 
-	private AtomicBoolean delayedStart = new AtomicBoolean(false);
-
 	public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) {
 		super(name);
 		this.metricRegistry = metricRegistry;
 	}
 
-	public void markDelayedStart() {
-		delayedStart.set(true);
-	}
-
-	@Override
-	public boolean isUp() {
-		return started.get() || delayedStart.get();
-	}
-
-	@Override
-	public boolean isOutOfService() {
-		return !started.get() && startException == null && !delayedStart.get();
-	}
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java

@@ -2,6 +2,8 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
+import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
+
 import com.codahale.metrics.Counter;
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
@@ -19,13 +21,15 @@ public class ConsumerInstrumentation extends Instrumentation {
 
 	public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) {
 		super(baseMetricName);
-		this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed"));
+
+		this.totalConsumed = registry
+				.counter(name(baseMetricName, Consumer.TOTAL_CONSUMED));
 		this.consumedPerSecond = registry
-				.meter(name(baseMetricName, "consumedPerSecond"));
+				.meter(name(baseMetricName, Consumer.CONSUMED_PER_SECOND));
 		this.totalConsumedFailures = registry
-				.counter(name(baseMetricName, "totalConsumedFailures"));
+				.counter(name(baseMetricName, Consumer.TOTAL_CONSUMED_FAILURES));
 		this.consumedFailuresPerSecond = registry
-				.meter(name(baseMetricName, "consumedFailuresPerSecond"));
+				.meter(name(baseMetricName, Consumer.CONSUMED_FAILURES_PER_SECOND));
 	}
 
 	public void markConsumed() {

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java

@@ -3,8 +3,12 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
+import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
+import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
+
 import com.codahale.metrics.MetricRegistry;
 
 /**
@@ -12,36 +16,32 @@ import com.codahale.metrics.MetricRegistry;
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
  */
 public class InstrumentationManager {
-	private final MetricRegistry metricRegistry;
-	private final Map<String, Object> runtime;
+
+	private final MetricRegistry metricRegistry = new MetricRegistry();
+	private final Map<String, Object> runtime = new ConcurrentHashMap<>();
+
 	private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>();
 	private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>();
 	private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>();
 
 	private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
 
-	public InstrumentationManager(MetricRegistry metricRegistry,
-			Map<String, Object> runtime) {
-		this.metricRegistry = metricRegistry;
-		this.runtime = runtime;
-	}
-
 	public ProducerInstrumentation getProducerInstrumentation(String destination) {
-		String key = "scs-rocketmq.producer." + destination;
+		String key = Producer.PREFIX + destination;
 		producerInstrumentations.putIfAbsent(key,
 				new ProducerInstrumentation(metricRegistry, key));
 		return producerInstrumentations.get(key);
 	}
 
 	public ConsumerInstrumentation getConsumerInstrumentation(String destination) {
-		String key = "scs-rocketmq.consumer." + destination;
+		String key = Consumer.PREFIX + destination;
 		consumeInstrumentations.putIfAbsent(key,
 				new ConsumerInstrumentation(metricRegistry, key));
 		return consumeInstrumentations.get(key);
 	}
 
 	public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) {
-		String key = "scs-rocketmq.consumerGroup." + group;
+		String key = Consumer.GROUP_PREFIX + group;
 		consumerGroupsInstrumentations.putIfAbsent(key,
 				new ConsumerGroupInstrumentation(metricRegistry, key));
 		return consumerGroupsInstrumentations.get(key);
@@ -59,4 +59,8 @@ public class InstrumentationManager {
 	public Map<String, Object> getRuntime() {
 		return runtime;
 	}
+
+	public MetricRegistry getMetricRegistry() {
+		return metricRegistry;
+	}
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java

@@ -2,6 +2,8 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
+import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
+
 import com.codahale.metrics.Counter;
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
@@ -19,12 +21,14 @@ public class ProducerInstrumentation extends Instrumentation {
 
 	public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) {
 		super(baseMetricName);
-		this.totalSent = registry.counter(name(baseMetricName, "totalSent"));
+
+		this.totalSent = registry.counter(name(baseMetricName, Producer.TOTAL_SENT));
 		this.totalSentFailures = registry
-				.counter(name(baseMetricName, "totalSentFailures"));
-		this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond"));
+				.counter(name(baseMetricName, Producer.TOTAL_SENT_FAILURES));
+		this.sentPerSecond = registry
+				.meter(name(baseMetricName, Producer.SENT_PER_SECOND));
 		this.sentFailuresPerSecond = registry
-				.meter(name(baseMetricName, "sentFailuresPerSecond"));
+				.meter(name(baseMetricName, Producer.SENT_FAILURES_PER_SECOND));
 	}
 
 	public void markSent() {

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java

@@ -9,7 +9,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 @ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
 public class RocketMQBinderConfigurationProperties {
 
-	private String namesrvAddr;
+	private String namesrvAddr = "127.0.0.1:9876";
 
 	private String logLevel = "ERROR";