Merge pull request #128 from fangjian0423/master

Refactor binder and update examples

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-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() {
+    public String getBar() {
-		return tag;
+        return bar;
     }
 
-	public void setTag(String tag) {
+    public void setBar(String bar) {
-		this.tag = tag;
+        this.bar = bar;
     }
 
     @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>
+            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>metrics-core</artifactId>
+            <artifactId>rocketmq-client</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.apache.rocketmq</groupId>
+            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>rocketmq-client</artifactId>
+            <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 org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
-import com.codahale.metrics.MetricRegistry;
 
 /**
  * @author Timur Valiev
@@ -18,23 +17,22 @@ import com.codahale.metrics.MetricRegistry;
 @Endpoint(id = ENDPOINT_ID)
 public class RocketMQBinderEndpoint {
 
-	private MetricRegistry metricRegistry = new MetricRegistry();
+	@Autowired(required = false)
-	private Map<String, Object> runtime = new ConcurrentHashMap<>();
+	private InstrumentationManager instrumentationManager;
 
 	@ReadOperation
 	public Map<String, Object> invoke() {
 		Map<String, Object> result = new HashMap<>();
-		result.put("metrics", metricRegistry().getMetrics());
+		if (instrumentationManager != null) {
-		result.put("runtime", runtime());
+			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,14 +12,12 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
  */
 public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
 
-	private final InstrumentationManager instrumentationManager;
+	@Autowired(required = false)
-
+	private InstrumentationManager instrumentationManager;
-	public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
-		this.instrumentationManager = instrumentationManager;
-	}
 
 	@Override
 	protected void doHealthCheck(Health.Builder builder) throws Exception {
+		if (instrumentationManager != null) {
 			if (instrumentationManager.getHealthInstrumentations().stream()
 					.allMatch(Instrumentation::isUp)) {
 				builder.up();
@@ -35,4 +34,11 @@ public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
 					.forEach(instrumentation1 -> builder
 							.withException(instrumentation1.getStartException()));
 		}
+		else {
+			builder.down();
+			builder.withDetail("warning",
+					"please add metrics-core dependency, we use it for metrics");
+		}
+
+	}
 }

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(
+	public ConsumersManager consumersManager() {
-			InstrumentationManager instrumentationManager) {
 		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(
+	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
-			InstrumentationManager instrumentationManager) {
+		return new RocketMQBinderHealthIndicator();
-		return new RocketMQBinderHealthIndicator(instrumentationManager);
 	}
 
 	@Bean
-	public InstrumentationManager instrumentationManager(
+	@ConditionalOnClass(name = "com.codahale.metrics.Counter")
-			RocketMQBinderEndpoint rocketBinderEndpoint) {
+	public InstrumentationManager instrumentationManager() {
-		return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(),
+		return new InstrumentationManager();
-				rocketBinderEndpoint.runtime());
 	}
 
 }

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
+
+		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
+
+		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
+		String tags = consumerProperties.getExtension().getTags();
-				: consumerProperties.getExtension().getTags();
+		Boolean isOrderly = consumerProperties.getExtension().getOrderly();
-		Boolean isOrderly = consumerProperties == null ? false
-				: consumerProperties.getExtension().getOrderly();
 
 		DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
 				destination, consumerProperties);
 
 		final CloudStreamMessageListener listener = isOrderly
-				? new CloudStreamMessageListenerOrderly(instrumentationManager)
+				? new CloudStreamMessageListenerOrderly()
-				: new CloudStreamMessageListenerConcurrently(instrumentationManager);
+				: 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
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-				.getConsumerInstrumentation(destination);
+			consumerInstrumentation = manager.getConsumerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
+			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
+					Optional.ofNullable(
-							.getConsumerInstrumentation(
+							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
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						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
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
+						.ifPresent(manager -> {
+							manager.getConsumerInstrumentation(
 									RocketMQInboundChannelAdapter.this.destination)
 									.markConsumedFailure();
+						});
 			}
 			else {
-				instrumentationManager
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						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
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-				.getProducerInstrumentation(destination);
+			producerInstrumentation = manager.getProducerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(producerInstrumentation);
+			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(
+			Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-					RocketMQBinderConstants.LASTSEND_TIMESTAMP,
+				manager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP,
 						Instant.now().toEpochMilli());
-			producerInstrumentation.markSent();
+			});
+			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"));
+				.counter(name(baseMetricName, Producer.TOTAL_SENT_FAILURES));
-		this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond"));
+		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";