Polish alibaba/spring-cloud-alibaba/#1283 : Renaming spring-cloud-starter-alibaba to be spring-cloud-alibaba-starters

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/pom.xml

@@ -0,0 +1,63 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.alibaba.cloud</groupId>
+        <artifactId>spring-cloud-alibaba-starters</artifactId>
+        <version>2.2.1.BUILD-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
+    <name>Spring Cloud Starter Stream RocketMQ</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-actuator</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-actuator-autoconfigure</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq;
+
+import static org.apache.rocketmq.spring.support.RocketMQHeaders.PREFIX;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ * @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
+ */
+public final class RocketMQBinderConstants {
+
+	/**
+	 * Header key for RocketMQ Transactional Args.
+	 */
+	public static final String ROCKET_TRANSACTIONAL_ARG = "TRANSACTIONAL_ARG";
+
+	/**
+	 * Default NameServer value.
+	 */
+	public static final String DEFAULT_NAME_SERVER = "127.0.0.1:9876";
+
+	/**
+	 * Default group for SCS RocketMQ Binder.
+	 */
+	public static final String DEFAULT_GROUP = PREFIX + "binder_default_group_name";
+
+	/**
+	 * RocketMQ re-consume times.
+	 */
+	public static final String ROCKETMQ_RECONSUME_TIMES = PREFIX + "RECONSUME_TIMES";
+
+	private RocketMQBinderConstants() {
+		throw new AssertionError("Must not instantiate constant utility class");
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQBinderUtils.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public final class RocketMQBinderUtils {
+
+	private RocketMQBinderUtils() {
+
+	}
+
+	public static RocketMQBinderConfigurationProperties mergeProperties(
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
+			RocketMQProperties rocketMQProperties) {
+		RocketMQBinderConfigurationProperties result = new RocketMQBinderConfigurationProperties();
+		if (StringUtils.isEmpty(rocketMQProperties.getNameServer())) {
+			result.setNameServer(rocketBinderConfigurationProperties.getNameServer());
+		}
+		else {
+			result.setNameServer(
+					Arrays.asList(rocketMQProperties.getNameServer().split(";")));
+		}
+		if (rocketMQProperties.getProducer() == null
+				|| StringUtils.isEmpty(rocketMQProperties.getProducer().getAccessKey())) {
+			result.setAccessKey(rocketBinderConfigurationProperties.getAccessKey());
+		}
+		else {
+			result.setAccessKey(rocketMQProperties.getProducer().getAccessKey());
+		}
+		if (rocketMQProperties.getProducer() == null
+				|| StringUtils.isEmpty(rocketMQProperties.getProducer().getSecretKey())) {
+			result.setSecretKey(rocketBinderConfigurationProperties.getSecretKey());
+		}
+		else {
+			result.setSecretKey(rocketMQProperties.getProducer().getSecretKey());
+		}
+		if (rocketMQProperties.getProducer() == null || StringUtils
+				.isEmpty(rocketMQProperties.getProducer().getCustomizedTraceTopic())) {
+			result.setCustomizedTraceTopic(
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+		}
+		else {
+			result.setCustomizedTraceTopic(
+					rocketMQProperties.getProducer().getCustomizedTraceTopic());
+		}
+		if (rocketMQProperties.getProducer() != null
+				&& rocketMQProperties.getProducer().isEnableMsgTrace()) {
+			result.setEnableMsgTrace(Boolean.TRUE);
+		}
+		else {
+			result.setEnableMsgTrace(
+					rocketBinderConfigurationProperties.isEnableMsgTrace());
+		}
+		return result;
+	}
+
+	public static String getNameServerStr(List<String> nameServerList) {
+		if (CollectionUtils.isEmpty(nameServerList)) {
+			return null;
+		}
+		return String.join(";", nameServerList);
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java

@@ -0,0 +1,327 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import com.alibaba.cloud.stream.binder.rocketmq.consuming.RocketMQListenerBindingContainer;
+import com.alibaba.cloud.stream.binder.rocketmq.integration.RocketMQInboundChannelAdapter;
+import com.alibaba.cloud.stream.binder.rocketmq.integration.RocketMQMessageHandler;
+import com.alibaba.cloud.stream.binder.rocketmq.integration.RocketMQMessageSource;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner;
+import com.alibaba.cloud.stream.binder.rocketmq.provisioning.selector.PartitionMessageQueueSelector;
+import com.alibaba.cloud.stream.binder.rocketmq.support.JacksonRocketMQHeaderMapper;
+import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
+
+import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.integration.StaticMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
+import org.springframework.integration.acks.AcknowledgmentCallback.Status;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQMessageChannelBinder extends
+		AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner>
+		implements
+		ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
+
+	private RocketMQExtendedBindingProperties extendedBindingProperties = new RocketMQExtendedBindingProperties();
+
+	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
+
+	private final RocketMQProperties rocketMQProperties;
+
+	private final InstrumentationManager instrumentationManager;
+
+	private Map<String, String> topicInUse = new HashMap<>();
+
+	public RocketMQMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider,
+			RocketMQExtendedBindingProperties extendedBindingProperties,
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
+			RocketMQProperties rocketMQProperties,
+			InstrumentationManager instrumentationManager) {
+		super(null, provisioningProvider);
+		this.extendedBindingProperties = extendedBindingProperties;
+		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
+		this.rocketMQProperties = rocketMQProperties;
+		this.instrumentationManager = instrumentationManager;
+	}
+
+	@Override
+	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+			ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
+			MessageChannel channel, MessageChannel errorChannel) throws Exception {
+		if (producerProperties.getExtension().getEnabled()) {
+
+			// if producerGroup is empty, using destination
+			String extendedProducerGroup = producerProperties.getExtension().getGroup();
+			String producerGroup = StringUtils.isEmpty(extendedProducerGroup)
+					? destination.getName() : extendedProducerGroup;
+
+			RocketMQBinderConfigurationProperties mergedProperties = RocketMQBinderUtils
+					.mergeProperties(rocketBinderConfigurationProperties,
+							rocketMQProperties);
+
+			RocketMQTemplate rocketMQTemplate;
+			if (producerProperties.getExtension().getTransactional()) {
+				Map<String, RocketMQTemplate> rocketMQTemplates = getBeanFactory()
+						.getBeansOfType(RocketMQTemplate.class);
+				if (rocketMQTemplates.size() == 0) {
+					throw new IllegalStateException(
+							"there is no RocketMQTemplate in Spring BeanFactory");
+				}
+				else if (rocketMQTemplates.size() > 1) {
+					throw new IllegalStateException(
+							"there is more than 1 RocketMQTemplates in Spring BeanFactory");
+				}
+				rocketMQTemplate = rocketMQTemplates.values().iterator().next();
+			}
+			else {
+				rocketMQTemplate = new RocketMQTemplate();
+				rocketMQTemplate.setObjectMapper(this.getApplicationContext()
+						.getBeansOfType(ObjectMapper.class).values().iterator().next());
+				DefaultMQProducer producer;
+				String ak = mergedProperties.getAccessKey();
+				String sk = mergedProperties.getSecretKey();
+				if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+					RPCHook rpcHook = new AclClientRPCHook(
+							new SessionCredentials(ak, sk));
+					producer = new DefaultMQProducer(producerGroup, rpcHook,
+							mergedProperties.isEnableMsgTrace(),
+							mergedProperties.getCustomizedTraceTopic());
+					producer.setVipChannelEnabled(false);
+					producer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook,
+							destination.getName() + "|" + UtilAll.getPid()));
+				}
+				else {
+					producer = new DefaultMQProducer(producerGroup);
+					producer.setVipChannelEnabled(
+							producerProperties.getExtension().getVipChannelEnabled());
+				}
+				producer.setNamesrvAddr(RocketMQBinderUtils
+						.getNameServerStr(mergedProperties.getNameServer()));
+				producer.setSendMsgTimeout(
+						producerProperties.getExtension().getSendMessageTimeout());
+				producer.setRetryTimesWhenSendFailed(
+						producerProperties.getExtension().getRetryTimesWhenSendFailed());
+				producer.setRetryTimesWhenSendAsyncFailed(producerProperties
+						.getExtension().getRetryTimesWhenSendAsyncFailed());
+				producer.setCompressMsgBodyOverHowmuch(producerProperties.getExtension()
+						.getCompressMessageBodyThreshold());
+				producer.setRetryAnotherBrokerWhenNotStoreOK(
+						producerProperties.getExtension().isRetryNextServer());
+				producer.setMaxMessageSize(
+						producerProperties.getExtension().getMaxMessageSize());
+				rocketMQTemplate.setProducer(producer);
+				if (producerProperties.isPartitioned()) {
+					rocketMQTemplate
+							.setMessageQueueSelector(new PartitionMessageQueueSelector());
+				}
+			}
+
+			RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
+					rocketMQTemplate, destination.getName(), producerGroup,
+					producerProperties.getExtension().getTransactional(),
+					instrumentationManager, producerProperties,
+					((AbstractMessageChannel) channel).getChannelInterceptors().stream()
+							.filter(channelInterceptor -> channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor)
+							.map(channelInterceptor -> ((MessageConverterConfigurer.PartitioningInterceptor) channelInterceptor))
+							.findFirst().orElse(null));
+			messageHandler.setBeanFactory(this.getApplicationContext().getBeanFactory());
+			messageHandler.setSync(producerProperties.getExtension().getSync());
+			messageHandler.setHeaderMapper(createHeaderMapper(producerProperties));
+			if (errorChannel != null) {
+				messageHandler.setSendFailureChannel(errorChannel);
+			}
+			return messageHandler;
+		}
+		else {
+			throw new RuntimeException("Binding for channel " + destination.getName()
+					+ " has been disabled, message can't be delivered");
+		}
+	}
+
+	@Override
+	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+			ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
+			MessageChannel errorChannel) throws Exception {
+		throw new UnsupportedOperationException(
+				"The abstract binder should not call this method");
+	}
+
+	@Override
+	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination,
+			String group,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties)
+			throws Exception {
+		if (group == null || "".equals(group)) {
+			throw new RuntimeException(
+					"'group must be configured for channel " + destination.getName());
+		}
+
+		RocketMQListenerBindingContainer listenerContainer = new RocketMQListenerBindingContainer(
+				consumerProperties, rocketBinderConfigurationProperties, this);
+		listenerContainer.setConsumerGroup(group);
+		listenerContainer.setTopic(destination.getName());
+		listenerContainer.setConsumeThreadMax(consumerProperties.getConcurrency());
+		listenerContainer.setSuspendCurrentQueueTimeMillis(
+				consumerProperties.getExtension().getSuspendCurrentQueueTimeMillis());
+		listenerContainer.setDelayLevelWhenNextConsume(
+				consumerProperties.getExtension().getDelayLevelWhenNextConsume());
+		listenerContainer
+				.setNameServer(rocketBinderConfigurationProperties.getNameServer());
+		listenerContainer.setHeaderMapper(createHeaderMapper(consumerProperties));
+
+		RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
+				listenerContainer, consumerProperties, instrumentationManager);
+
+		topicInUse.put(destination.getName(), group);
+
+		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
+				group, consumerProperties);
+		if (consumerProperties.getMaxAttempts() > 1) {
+			rocketInboundChannelAdapter
+					.setRetryTemplate(buildRetryTemplate(consumerProperties));
+			rocketInboundChannelAdapter
+					.setRecoveryCallback(errorInfrastructure.getRecoverer());
+		}
+		else {
+			rocketInboundChannelAdapter
+					.setErrorChannel(errorInfrastructure.getErrorChannel());
+		}
+
+		return rocketInboundChannelAdapter;
+	}
+
+	@Override
+	protected PolledConsumerResources createPolledConsumerResources(String name,
+			String group, ConsumerDestination destination,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
+		RocketMQMessageSource rocketMQMessageSource = new RocketMQMessageSource(
+				rocketBinderConfigurationProperties, consumerProperties, name, group);
+		return new PolledConsumerResources(rocketMQMessageSource,
+				registerErrorInfrastructure(destination, group, consumerProperties,
+						true));
+	}
+
+	@Override
+	protected MessageHandler getPolledConsumerErrorMessageHandler(
+			ConsumerDestination destination, String group,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> properties) {
+		return message -> {
+			if (message.getPayload() instanceof MessagingException) {
+				AcknowledgmentCallback ack = StaticMessageHeaderAccessor
+						.getAcknowledgmentCallback(
+								((MessagingException) message.getPayload())
+										.getFailedMessage());
+				if (ack != null) {
+					if (properties.getExtension().shouldRequeue()) {
+						ack.acknowledge(Status.REQUEUE);
+					}
+					else {
+						ack.acknowledge(Status.REJECT);
+					}
+				}
+			}
+		};
+	}
+
+	@Override
+	public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedConsumerProperties(channelName);
+	}
+
+	@Override
+	public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedProducerProperties(channelName);
+	}
+
+	public Map<String, String> getTopicInUse() {
+		return topicInUse;
+	}
+
+	@Override
+	public String getDefaultsPrefix() {
+		return extendedBindingProperties.getDefaultsPrefix();
+	}
+
+	@Override
+	public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
+		return extendedBindingProperties.getExtendedPropertiesEntryClass();
+	}
+
+	public void setExtendedBindingProperties(
+			RocketMQExtendedBindingProperties extendedBindingProperties) {
+		this.extendedBindingProperties = extendedBindingProperties;
+	}
+
+	private RocketMQHeaderMapper createHeaderMapper(
+			final ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties) {
+		Set<String> trustedPackages = extendedConsumerProperties.getExtension()
+				.getTrustedPackages();
+		return createHeaderMapper(trustedPackages);
+	}
+
+	private RocketMQHeaderMapper createHeaderMapper(
+			final ExtendedProducerProperties<RocketMQProducerProperties> producerProperties) {
+		return createHeaderMapper(Collections.emptyList());
+	}
+
+	private RocketMQHeaderMapper createHeaderMapper(Collection<String> trustedPackages) {
+		ObjectMapper objectMapper = this.getApplicationContext()
+				.getBeansOfType(ObjectMapper.class).values().iterator().next();
+		JacksonRocketMQHeaderMapper headerMapper = new JacksonRocketMQHeaderMapper(
+				objectMapper);
+		if (!StringUtils.isEmpty(trustedPackages)) {
+			headerMapper.addTrustedPackages(trustedPackages);
+		}
+		return headerMapper;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.actuator;
+
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.Instrumentation;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.health.AbstractHealthIndicator;
+import org.springframework.boot.actuate.health.Health;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
+
+	@Autowired
+	private InstrumentationManager instrumentationManager;
+
+	@Override
+	protected void doHealthCheck(Health.Builder builder) throws Exception {
+		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()));
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.config;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@Configuration(proxyBeanMethods = false)
+@Import({ RocketMQAutoConfiguration.class,
+		RocketMQBinderHealthIndicatorAutoConfiguration.class })
+@EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
+		RocketMQExtendedBindingProperties.class })
+public class RocketMQBinderAutoConfiguration {
+
+	private final RocketMQExtendedBindingProperties extendedBindingProperties;
+
+	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
+
+	@Autowired(required = false)
+	private RocketMQProperties rocketMQProperties = new RocketMQProperties();
+
+	@Autowired
+	public RocketMQBinderAutoConfiguration(
+			RocketMQExtendedBindingProperties extendedBindingProperties,
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
+		this.extendedBindingProperties = extendedBindingProperties;
+		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
+	}
+
+	@Bean
+	public RocketMQTopicProvisioner provisioningProvider() {
+		return new RocketMQTopicProvisioner();
+	}
+
+	@Bean
+	public RocketMQMessageChannelBinder rocketMessageChannelBinder(
+			RocketMQTopicProvisioner provisioningProvider,
+			InstrumentationManager instrumentationManager) {
+		RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
+				provisioningProvider, extendedBindingProperties,
+				rocketBinderConfigurationProperties, rocketMQProperties,
+				instrumentationManager);
+		binder.setExtendedBindingProperties(extendedBindingProperties);
+		return binder;
+	}
+
+	@Bean
+	public InstrumentationManager instrumentationManager() {
+		return new InstrumentationManager();
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/config/RocketMQBinderHealthIndicatorAutoConfiguration.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.config;
+
+import com.alibaba.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator;
+
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnClass(Endpoint.class)
+public class RocketMQBinderHealthIndicatorAutoConfiguration {
+
+	@Bean
+	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
+		return new RocketMQBinderHealthIndicator();
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.config;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
+import org.apache.rocketmq.spring.config.RocketMQTransactionAnnotationProcessor;
+import org.apache.rocketmq.spring.config.TransactionHandlerRegistry;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@Configuration(proxyBeanMethods = false)
+@AutoConfigureAfter(RocketMQAutoConfiguration.class)
+@ConditionalOnMissingBean(DefaultMQProducer.class)
+public class RocketMQComponent4BinderAutoConfiguration {
+
+	private final Environment environment;
+
+	public RocketMQComponent4BinderAutoConfiguration(Environment environment) {
+		this.environment = environment;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(DefaultMQProducer.class)
+	public DefaultMQProducer defaultMQProducer() {
+		DefaultMQProducer producer;
+		String configNameServer = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.name-server:${rocketmq.producer.name-server:}}");
+		String ak = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.access-key:${rocketmq.producer.access-key:}}");
+		String sk = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.secret-key:${rocketmq.producer.secret-key:}}");
+		if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+			producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP,
+					new AclClientRPCHook(new SessionCredentials(ak, sk)));
+			producer.setVipChannelEnabled(false);
+		}
+		else {
+			producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP);
+		}
+		if (StringUtils.isEmpty(configNameServer)) {
+			configNameServer = RocketMQBinderConstants.DEFAULT_NAME_SERVER;
+		}
+		producer.setNamesrvAddr(configNameServer);
+		return producer;
+	}
+
+	@Bean(destroyMethod = "destroy")
+	@ConditionalOnMissingBean
+	public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
+			ObjectMapper objectMapper) {
+		RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
+		rocketMQTemplate.setProducer(mqProducer);
+		rocketMQTemplate.setObjectMapper(objectMapper);
+		return rocketMQTemplate;
+	}
+
+	@Bean
+	@ConditionalOnBean(RocketMQTemplate.class)
+	@ConditionalOnMissingBean(TransactionHandlerRegistry.class)
+	public TransactionHandlerRegistry transactionHandlerRegistry(
+			RocketMQTemplate template) {
+		return new TransactionHandlerRegistry(template);
+	}
+
+	@Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME)
+	@ConditionalOnBean(TransactionHandlerRegistry.class)
+	public static RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(
+			TransactionHandlerRegistry transactionHandlerRegistry) {
+		return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java

@@ -0,0 +1,465 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.consuming;
+
+import java.util.List;
+import java.util.Objects;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderUtils;
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.MessageModel;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import static com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKETMQ_RECONSUME_TIMES;
+
+/**
+ * A class that Listen on rocketmq message.
+ * <p>
+ * this class will delegate {@link RocketMQListener} to handle message
+ *
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ * @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
+ * @see RocketMQListener
+ */
+public class RocketMQListenerBindingContainer
+		implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
+
+	private final static Logger log = LoggerFactory
+			.getLogger(RocketMQListenerBindingContainer.class);
+
+	private long suspendCurrentQueueTimeMillis = 1000;
+
+	/**
+	 * Message consume retry strategy<br>
+	 * -1,no retry,put into DLQ directly<br>
+	 * 0,broker control retry frequency<br>
+	 * >0,client control retry frequency.
+	 */
+	private int delayLevelWhenNextConsume = 0;
+
+	private List<String> nameServer;
+
+	private String consumerGroup;
+
+	private String topic;
+
+	private int consumeThreadMax = 64;
+
+	private String charset = "UTF-8";
+
+	private RocketMQListener rocketMQListener;
+
+	private RocketMQHeaderMapper headerMapper;
+
+	private DefaultMQPushConsumer consumer;
+
+	private boolean running;
+
+	private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
+
+	private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
+
+	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
+
+	// The following properties came from RocketMQConsumerProperties.
+	private ConsumeMode consumeMode;
+
+	private SelectorType selectorType;
+
+	private String selectorExpression;
+
+	private MessageModel messageModel;
+
+	public RocketMQListenerBindingContainer(
+			ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
+			RocketMQMessageChannelBinder rocketMQMessageChannelBinder) {
+		this.rocketMQConsumerProperties = rocketMQConsumerProperties;
+		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
+		this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder;
+		this.consumeMode = rocketMQConsumerProperties.getExtension().getOrderly()
+				? ConsumeMode.ORDERLY : ConsumeMode.CONCURRENTLY;
+		if (StringUtils.isEmpty(rocketMQConsumerProperties.getExtension().getSql())) {
+			this.selectorType = SelectorType.TAG;
+			this.selectorExpression = rocketMQConsumerProperties.getExtension().getTags();
+		}
+		else {
+			this.selectorType = SelectorType.SQL92;
+			this.selectorExpression = rocketMQConsumerProperties.getExtension().getSql();
+		}
+		this.messageModel = rocketMQConsumerProperties.getExtension().getBroadcasting()
+				? MessageModel.BROADCASTING : MessageModel.CLUSTERING;
+	}
+
+	@Override
+	public void setupMessageListener(RocketMQListener<?> rocketMQListener) {
+		this.rocketMQListener = rocketMQListener;
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		this.setRunning(false);
+		if (Objects.nonNull(consumer)) {
+			consumer.shutdown();
+		}
+		log.info("container destroyed, {}", this.toString());
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		initRocketMQPushConsumer();
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return true;
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	@Override
+	public void start() {
+		if (this.isRunning()) {
+			throw new IllegalStateException(
+					"container already running. " + this.toString());
+		}
+
+		try {
+			consumer.start();
+		}
+		catch (MQClientException e) {
+			throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
+		}
+		this.setRunning(true);
+
+		log.info("running container: {}", this.toString());
+	}
+
+	@Override
+	public void stop() {
+		if (this.isRunning()) {
+			if (Objects.nonNull(consumer)) {
+				consumer.shutdown();
+			}
+			setRunning(false);
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return running;
+	}
+
+	private void setRunning(boolean running) {
+		this.running = running;
+	}
+
+	@Override
+	public int getPhase() {
+		return Integer.MAX_VALUE;
+	}
+
+	private void initRocketMQPushConsumer() throws MQClientException {
+		Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
+		Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
+		Assert.notNull(nameServer, "Property 'nameServer' is required");
+		Assert.notNull(topic, "Property 'topic' is required");
+
+		String ak = rocketBinderConfigurationProperties.getAccessKey();
+		String sk = rocketBinderConfigurationProperties.getSecretKey();
+		if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+			RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk));
+			consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook,
+					new AllocateMessageQueueAveragely(),
+					rocketBinderConfigurationProperties.isEnableMsgTrace(),
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+			consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook,
+					topic + "|" + UtilAll.getPid()));
+			consumer.setVipChannelEnabled(false);
+		}
+		else {
+			consumer = new DefaultMQPushConsumer(consumerGroup,
+					rocketBinderConfigurationProperties.isEnableMsgTrace(),
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+		}
+
+		consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(nameServer));
+		consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
+		consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());
+
+		switch (messageModel) {
+		case BROADCASTING:
+			consumer.setMessageModel(
+					org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
+			break;
+		case CLUSTERING:
+			consumer.setMessageModel(
+					org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
+			break;
+		default:
+			throw new IllegalArgumentException("Property 'messageModel' was wrong.");
+		}
+
+		switch (selectorType) {
+		case TAG:
+			consumer.subscribe(topic, selectorExpression);
+			break;
+		case SQL92:
+			consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
+			break;
+		default:
+			throw new IllegalArgumentException("Property 'selectorType' was wrong.");
+		}
+
+		switch (consumeMode) {
+		case ORDERLY:
+			consumer.setMessageListener(new DefaultMessageListenerOrderly());
+			break;
+		case CONCURRENTLY:
+			consumer.setMessageListener(new DefaultMessageListenerConcurrently());
+			break;
+		default:
+			throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
+		}
+
+		if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
+			((RocketMQPushConsumerLifecycleListener) rocketMQListener)
+					.prepareStart(consumer);
+		}
+
+	}
+
+	@Override
+	public String toString() {
+		return "RocketMQListenerBindingContainer{" + "consumerGroup='" + consumerGroup
+				+ '\'' + ", nameServer='" + nameServer + '\'' + ", topic='" + topic + '\''
+				+ ", consumeMode=" + consumeMode + ", selectorType=" + selectorType
+				+ ", selectorExpression='" + selectorExpression + '\'' + ", messageModel="
+				+ messageModel + '}';
+	}
+
+	public long getSuspendCurrentQueueTimeMillis() {
+		return suspendCurrentQueueTimeMillis;
+	}
+
+	public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
+		this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
+	}
+
+	public int getDelayLevelWhenNextConsume() {
+		return delayLevelWhenNextConsume;
+	}
+
+	public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
+		this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
+	}
+
+	public List<String> getNameServer() {
+		return nameServer;
+	}
+
+	public void setNameServer(List<String> nameServer) {
+		this.nameServer = nameServer;
+	}
+
+	public String getConsumerGroup() {
+		return consumerGroup;
+	}
+
+	public void setConsumerGroup(String consumerGroup) {
+		this.consumerGroup = consumerGroup;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+
+	public int getConsumeThreadMax() {
+		return consumeThreadMax;
+	}
+
+	public void setConsumeThreadMax(int consumeThreadMax) {
+		this.consumeThreadMax = consumeThreadMax;
+	}
+
+	public String getCharset() {
+		return charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public RocketMQListener getRocketMQListener() {
+		return rocketMQListener;
+	}
+
+	public void setRocketMQListener(RocketMQListener rocketMQListener) {
+		this.rocketMQListener = rocketMQListener;
+	}
+
+	public DefaultMQPushConsumer getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(DefaultMQPushConsumer consumer) {
+		this.consumer = consumer;
+	}
+
+	public ExtendedConsumerProperties<RocketMQConsumerProperties> getRocketMQConsumerProperties() {
+		return rocketMQConsumerProperties;
+	}
+
+	public ConsumeMode getConsumeMode() {
+		return consumeMode;
+	}
+
+	public SelectorType getSelectorType() {
+		return selectorType;
+	}
+
+	public String getSelectorExpression() {
+		return selectorExpression;
+	}
+
+	public MessageModel getMessageModel() {
+		return messageModel;
+	}
+
+	public RocketMQHeaderMapper getHeaderMapper() {
+		return headerMapper;
+	}
+
+	public void setHeaderMapper(RocketMQHeaderMapper headerMapper) {
+		this.headerMapper = headerMapper;
+	}
+
+	/**
+	 * Convert rocketmq {@link MessageExt} to Spring {@link Message}.
+	 * @param messageExt the rocketmq message
+	 * @return the converted Spring {@link Message}
+	 */
+	@SuppressWarnings("unchecked")
+	private Message convertToSpringMessage(MessageExt messageExt) {
+
+		// add reconsume-times header to messageExt
+		int reconsumeTimes = messageExt.getReconsumeTimes();
+		messageExt.putUserProperty(ROCKETMQ_RECONSUME_TIMES,
+				String.valueOf(reconsumeTimes));
+		Message message = RocketMQUtil.convertToSpringMessage(messageExt);
+		return MessageBuilder.fromMessage(message)
+				.copyHeaders(headerMapper.toHeaders(messageExt.getProperties())).build();
+	}
+
+	public class DefaultMessageListenerConcurrently
+			implements MessageListenerConcurrently {
+
+		@SuppressWarnings({ "unchecked", "Duplicates" })
+		@Override
+		public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+				ConsumeConcurrentlyContext context) {
+			for (MessageExt messageExt : msgs) {
+				log.debug("received msg: {}", messageExt);
+				try {
+					long now = System.currentTimeMillis();
+					rocketMQListener.onMessage(convertToSpringMessage(messageExt));
+					long costTime = System.currentTimeMillis() - now;
+					log.debug("consume {} message key:[{}] cost: {} ms",
+							messageExt.getMsgId(), messageExt.getKeys(), costTime);
+				}
+				catch (Exception e) {
+					log.warn("consume message failed. messageExt:{}", messageExt, e);
+					context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
+					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+				}
+			}
+
+			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+		}
+
+	}
+
+	public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
+
+		@SuppressWarnings({ "unchecked", "Duplicates" })
+		@Override
+		public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
+				ConsumeOrderlyContext context) {
+			for (MessageExt messageExt : msgs) {
+				log.debug("received msg: {}", messageExt);
+				try {
+					long now = System.currentTimeMillis();
+					rocketMQListener.onMessage(convertToSpringMessage(messageExt));
+					long costTime = System.currentTimeMillis() - now;
+					log.info("consume {} message key:[{}] cost: {} ms",
+							messageExt.getMsgId(), messageExt.getKeys(), costTime);
+				}
+				catch (Exception e) {
+					log.warn("consume message failed. messageExt:{}", messageExt, e);
+					context.setSuspendCurrentQueueTimeMillis(
+							suspendCurrentQueueTimeMillis);
+					return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
+				}
+			}
+
+			return ConsumeOrderlyStatus.SUCCESS;
+		}
+
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/consuming/RocketMQMessageQueueChooser.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.consuming;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQMessageQueueChooser {
+
+	private volatile int queueIndex = 0;
+
+	private volatile List<MessageQueue> messageQueues;
+
+	public MessageQueue choose() {
+		return messageQueues.get(queueIndex);
+	}
+
+	public int requeue() {
+		if (queueIndex - 1 < 0) {
+			this.queueIndex = messageQueues.size() - 1;
+		}
+		else {
+			this.queueIndex = this.queueIndex - 1;
+		}
+		return this.queueIndex;
+	}
+
+	public void increment() {
+		this.queueIndex = (this.queueIndex + 1) % messageQueues.size();
+	}
+
+	public void reset(Set<MessageQueue> queueSet) {
+		this.messageQueues = null;
+		this.messageQueues = new ArrayList<>(queueSet);
+		this.queueIndex = 0;
+	}
+
+	public List<MessageQueue> getMessageQueues() {
+		return messageQueues;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.integration;
+
+import com.alibaba.cloud.stream.binder.rocketmq.consuming.RocketMQListenerBindingContainer;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.Instrumentation;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.RetryListener;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
+
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQInboundChannelAdapter.class);
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<? extends Object> recoveryCallback;
+
+	private RocketMQListenerBindingContainer rocketMQListenerContainer;
+
+	private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
+
+	private final InstrumentationManager instrumentationManager;
+
+	public RocketMQInboundChannelAdapter(
+			RocketMQListenerBindingContainer rocketMQListenerContainer,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
+			InstrumentationManager instrumentationManager) {
+		this.rocketMQListenerContainer = rocketMQListenerContainer;
+		this.consumerProperties = consumerProperties;
+		this.instrumentationManager = instrumentationManager;
+	}
+
+	@Override
+	protected void onInit() {
+		if (consumerProperties == null
+				|| !consumerProperties.getExtension().getEnabled()) {
+			return;
+		}
+		super.onInit();
+		if (this.retryTemplate != null) {
+			Assert.state(getErrorChannel() == null,
+					"Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+							+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+							+ "send an error message when retries are exhausted");
+		}
+
+		BindingRocketMQListener listener = new BindingRocketMQListener();
+		rocketMQListenerContainer.setRocketMQListener(listener);
+
+		if (retryTemplate != null) {
+			this.retryTemplate.registerListener(listener);
+		}
+
+		try {
+			rocketMQListenerContainer.afterPropertiesSet();
+
+		}
+		catch (Exception e) {
+			log.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
+			throw new IllegalArgumentException(
+					"rocketMQListenerContainer init error: " + e.getMessage(), e);
+		}
+
+		instrumentationManager.addHealthInstrumentation(
+				new Instrumentation(rocketMQListenerContainer.getTopic()
+						+ rocketMQListenerContainer.getConsumerGroup()));
+	}
+
+	@Override
+	protected void doStart() {
+		if (consumerProperties == null
+				|| !consumerProperties.getExtension().getEnabled()) {
+			return;
+		}
+		try {
+			rocketMQListenerContainer.start();
+			instrumentationManager
+					.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
+							+ rocketMQListenerContainer.getConsumerGroup())
+					.markStartedSuccessfully();
+		}
+		catch (Exception e) {
+			instrumentationManager
+					.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
+							+ rocketMQListenerContainer.getConsumerGroup())
+					.markStartFailed(e);
+			log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
+			throw new MessagingException(MessageBuilder.withPayload(
+					"RocketMQTemplate startup failed, Caused by " + e.getMessage())
+					.build(), e);
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		rocketMQListenerContainer.stop();
+	}
+
+	public void setRetryTemplate(RetryTemplate retryTemplate) {
+		this.retryTemplate = retryTemplate;
+	}
+
+	public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
+		this.recoveryCallback = recoveryCallback;
+	}
+
+	protected class BindingRocketMQListener
+			implements RocketMQListener<Message>, RetryListener {
+
+		@Override
+		public void onMessage(Message message) {
+			boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
+			if (enableRetry) {
+				RocketMQInboundChannelAdapter.this.retryTemplate.execute(context -> {
+					RocketMQInboundChannelAdapter.this.sendMessage(message);
+					return null;
+				}, (RecoveryCallback<Object>) RocketMQInboundChannelAdapter.this.recoveryCallback);
+			}
+			else {
+				RocketMQInboundChannelAdapter.this.sendMessage(message);
+			}
+		}
+
+		@Override
+		public <T, E extends Throwable> boolean open(RetryContext context,
+				RetryCallback<T, E> callback) {
+			return true;
+		}
+
+		@Override
+		public <T, E extends Throwable> void close(RetryContext context,
+				RetryCallback<T, E> callback, Throwable throwable) {
+		}
+
+		@Override
+		public <T, E extends Throwable> void onError(RetryContext context,
+				RetryCallback<T, E> callback, Throwable throwable) {
+
+		}
+
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java

@@ -0,0 +1,302 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.integration;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.Instrumentation;
+import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
+
+	private final static Logger log = LoggerFactory
+			.getLogger(RocketMQMessageHandler.class);
+
+	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
+
+	private MessageChannel sendFailureChannel;
+
+	private final RocketMQTemplate rocketMQTemplate;
+
+	private RocketMQHeaderMapper headerMapper;
+
+	private final Boolean transactional;
+
+	private final String destination;
+
+	private final String groupName;
+
+	private final InstrumentationManager instrumentationManager;
+
+	private boolean sync = false;
+
+	private volatile boolean running = false;
+
+	private ExtendedProducerProperties<RocketMQProducerProperties> producerProperties;
+
+	private MessageConverterConfigurer.PartitioningInterceptor partitioningInterceptor;
+
+	public RocketMQMessageHandler(RocketMQTemplate rocketMQTemplate, String destination,
+			String groupName, Boolean transactional,
+			InstrumentationManager instrumentationManager,
+			ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
+			MessageConverterConfigurer.PartitioningInterceptor partitioningInterceptor) {
+		this.rocketMQTemplate = rocketMQTemplate;
+		this.destination = destination;
+		this.groupName = groupName;
+		this.transactional = transactional;
+		this.instrumentationManager = instrumentationManager;
+		this.producerProperties = producerProperties;
+		this.partitioningInterceptor = partitioningInterceptor;
+	}
+
+	@Override
+	public void start() {
+		if (!transactional) {
+			instrumentationManager
+					.addHealthInstrumentation(new Instrumentation(destination));
+			try {
+				rocketMQTemplate.afterPropertiesSet();
+				instrumentationManager.getHealthInstrumentation(destination)
+						.markStartedSuccessfully();
+			}
+			catch (Exception e) {
+				instrumentationManager.getHealthInstrumentation(destination)
+						.markStartFailed(e);
+				log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
+				throw new MessagingException(MessageBuilder.withPayload(
+						"RocketMQTemplate startup failed, Caused by " + e.getMessage())
+						.build(), e);
+			}
+		}
+		if (producerProperties.isPartitioned()) {
+			try {
+				List<MessageQueue> messageQueues = rocketMQTemplate.getProducer()
+						.fetchPublishMessageQueues(destination);
+				if (producerProperties.getPartitionCount() != messageQueues.size()) {
+					logger.info(String.format(
+							"The partition count of topic '%s' will change from '%s' to '%s'",
+							destination, producerProperties.getPartitionCount(),
+							messageQueues.size()));
+					producerProperties.setPartitionCount(messageQueues.size());
+					partitioningInterceptor
+							.setPartitionCount(producerProperties.getPartitionCount());
+				}
+			}
+			catch (MQClientException e) {
+				logger.error("fetch publish message queues fail", e);
+			}
+		}
+		running = true;
+	}
+
+	@Override
+	public void stop() {
+		if (!transactional) {
+			rocketMQTemplate.destroy();
+		}
+		running = false;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return running;
+	}
+
+	@Override
+	protected void handleMessageInternal(
+			org.springframework.messaging.Message<?> message) {
+		try {
+			// issue 737 fix
+			Map<String, String> jsonHeaders = headerMapper
+					.fromHeaders(message.getHeaders());
+			message = org.springframework.messaging.support.MessageBuilder
+					.fromMessage(message).copyHeaders(jsonHeaders).build();
+
+			final StringBuilder topicWithTags = new StringBuilder(destination);
+			String tags = Optional
+					.ofNullable(message.getHeaders().get(RocketMQHeaders.TAGS)).orElse("")
+					.toString();
+			if (!StringUtils.isEmpty(tags)) {
+				topicWithTags.append(":").append(tags);
+			}
+
+			SendResult sendRes = null;
+			if (transactional) {
+				sendRes = rocketMQTemplate.sendMessageInTransaction(groupName,
+						topicWithTags.toString(), message, message.getHeaders()
+								.get(RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG));
+				log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
+			}
+			else {
+				int delayLevel = 0;
+				try {
+					Object delayLevelObj = message.getHeaders()
+							.getOrDefault(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 0);
+					if (delayLevelObj instanceof Number) {
+						delayLevel = ((Number) delayLevelObj).intValue();
+					}
+					else if (delayLevelObj instanceof String) {
+						delayLevel = Integer.parseInt((String) delayLevelObj);
+					}
+				}
+				catch (Exception e) {
+					// ignore
+				}
+				boolean needSelectQueue = message.getHeaders()
+						.containsKey(BinderHeaders.PARTITION_HEADER);
+				if (sync) {
+					if (needSelectQueue) {
+						sendRes = rocketMQTemplate.syncSendOrderly(
+								topicWithTags.toString(), message, "",
+								rocketMQTemplate.getProducer().getSendMsgTimeout());
+					}
+					else {
+						sendRes = rocketMQTemplate.syncSend(topicWithTags.toString(),
+								message,
+								rocketMQTemplate.getProducer().getSendMsgTimeout(),
+								delayLevel);
+					}
+					log.debug("sync send to topic " + topicWithTags + " " + sendRes);
+				}
+				else {
+					Message<?> finalMessage = message;
+					SendCallback sendCallback = new SendCallback() {
+						@Override
+						public void onSuccess(SendResult sendResult) {
+							log.debug("async send to topic " + topicWithTags + " "
+									+ sendResult);
+						}
+
+						@Override
+						public void onException(Throwable e) {
+							log.error("RocketMQ Message hasn't been sent. Caused by "
+									+ e.getMessage());
+							if (getSendFailureChannel() != null) {
+								getSendFailureChannel().send(
+										RocketMQMessageHandler.this.errorMessageStrategy
+												.buildErrorMessage(new MessagingException(
+														finalMessage, e), null));
+							}
+						}
+					};
+					if (needSelectQueue) {
+						rocketMQTemplate.asyncSendOrderly(topicWithTags.toString(),
+								message, "", sendCallback,
+								rocketMQTemplate.getProducer().getSendMsgTimeout());
+					}
+					else {
+						rocketMQTemplate.asyncSend(topicWithTags.toString(), message,
+								sendCallback);
+					}
+				}
+			}
+			if (sendRes != null && !sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
+				if (getSendFailureChannel() != null) {
+					this.getSendFailureChannel().send(message);
+				}
+				else {
+					throw new MessagingException(message,
+							new MQClientException("message hasn't been sent", null));
+				}
+			}
+		}
+		catch (Exception e) {
+			log.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
+			if (getSendFailureChannel() != null) {
+				getSendFailureChannel().send(this.errorMessageStrategy
+						.buildErrorMessage(new MessagingException(message, e), null));
+			}
+			else {
+				throw new MessagingException(message, e);
+			}
+		}
+
+	}
+
+	/**
+	 * Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent
+	 * to this channel with a payload of a {@link MessagingException} with the failed
+	 * message and cause.
+	 * @param sendFailureChannel the failure channel.
+	 * @since 0.2.2
+	 */
+	public void setSendFailureChannel(MessageChannel sendFailureChannel) {
+		this.sendFailureChannel = sendFailureChannel;
+	}
+
+	/**
+	 * Set the error message strategy implementation to use when sending error messages
+	 * after send failures. Cannot be null.
+	 * @param errorMessageStrategy the implementation.
+	 * @since 0.2.2
+	 */
+	public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
+		Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null");
+		this.errorMessageStrategy = errorMessageStrategy;
+	}
+
+	public MessageChannel getSendFailureChannel() {
+		return sendFailureChannel;
+	}
+
+	public void setSync(boolean sync) {
+		this.sync = sync;
+	}
+
+	public RocketMQHeaderMapper getHeaderMapper() {
+		return headerMapper;
+	}
+
+	public void setHeaderMapper(RocketMQHeaderMapper headerMapper) {
+		this.headerMapper = headerMapper;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/RocketMQMessageSource.java

@@ -0,0 +1,382 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.integration;
+
+import java.util.List;
+import java.util.Set;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderUtils;
+import com.alibaba.cloud.stream.binder.rocketmq.consuming.RocketMQMessageQueueChooser;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.MessageQueueListener;
+import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
+import org.springframework.integration.acks.AcknowledgmentCallbackFactory;
+import org.springframework.integration.endpoint.AbstractMessageSource;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQMessageSource extends AbstractMessageSource<Object>
+		implements DisposableBean, Lifecycle {
+
+	private final static Logger log = LoggerFactory
+			.getLogger(RocketMQMessageSource.class);
+
+	private final RocketMQCallbackFactory ackCallbackFactory;
+
+	private final RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties;
+
+	private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
+
+	private final String topic;
+
+	private final String group;
+
+	private final Object consumerMonitor = new Object();
+
+	private DefaultMQPullConsumer consumer;
+
+	private boolean running;
+
+	private MessageSelector messageSelector;
+
+	private RocketMQMessageQueueChooser messageQueueChooser = new RocketMQMessageQueueChooser();
+
+	public RocketMQMessageSource(
+			RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
+			String topic, String group) {
+		this(new RocketMQCallbackFactory(), rocketMQBinderConfigurationProperties,
+				rocketMQConsumerProperties, topic, group);
+	}
+
+	public RocketMQMessageSource(RocketMQCallbackFactory ackCallbackFactory,
+			RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
+			String topic, String group) {
+		this.ackCallbackFactory = ackCallbackFactory;
+		this.rocketMQBinderConfigurationProperties = rocketMQBinderConfigurationProperties;
+		this.rocketMQConsumerProperties = rocketMQConsumerProperties;
+		this.topic = topic;
+		this.group = group;
+	}
+
+	@Override
+	public synchronized void start() {
+		if (this.isRunning()) {
+			throw new IllegalStateException(
+					"pull consumer already running. " + this.toString());
+		}
+		try {
+			consumer = new DefaultMQPullConsumer(group);
+			consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(
+					rocketMQBinderConfigurationProperties.getNameServer()));
+			consumer.setConsumerPullTimeoutMillis(
+					rocketMQConsumerProperties.getExtension().getPullTimeout());
+			consumer.setMessageModel(MessageModel.CLUSTERING);
+
+			String tags = rocketMQConsumerProperties.getExtension().getTags();
+			String sql = rocketMQConsumerProperties.getExtension().getSql();
+
+			if (!StringUtils.isEmpty(tags) && !StringUtils.isEmpty(sql)) {
+				messageSelector = MessageSelector.byTag(tags);
+			}
+			else if (!StringUtils.isEmpty(tags)) {
+				messageSelector = MessageSelector.byTag(tags);
+			}
+			else if (!StringUtils.isEmpty(sql)) {
+				messageSelector = MessageSelector.bySql(sql);
+			}
+
+			consumer.registerMessageQueueListener(topic, new MessageQueueListener() {
+				@Override
+				public void messageQueueChanged(String topic, Set<MessageQueue> mqAll,
+						Set<MessageQueue> mqDivided) {
+					log.info(
+							"messageQueueChanged, topic='{}', mqAll=`{}`, mqDivided=`{}`",
+							topic, mqAll, mqDivided);
+					switch (consumer.getMessageModel()) {
+					case BROADCASTING:
+						RocketMQMessageSource.this.resetMessageQueues(mqAll);
+						break;
+					case CLUSTERING:
+						RocketMQMessageSource.this.resetMessageQueues(mqDivided);
+						break;
+					default:
+						break;
+					}
+				}
+			});
+			consumer.start();
+		}
+		catch (MQClientException e) {
+			log.error("DefaultMQPullConsumer startup error: " + e.getMessage(), e);
+		}
+		this.setRunning(true);
+	}
+
+	@Override
+	public synchronized void stop() {
+		if (this.isRunning()) {
+			this.setRunning(false);
+			consumer.shutdown();
+		}
+	}
+
+	@Override
+	public synchronized boolean isRunning() {
+		return running;
+	}
+
+	@Override
+	protected synchronized Object doReceive() {
+		if (messageQueueChooser.getMessageQueues() == null
+				|| messageQueueChooser.getMessageQueues().size() == 0) {
+			return null;
+		}
+		try {
+			int count = 0;
+			while (count < messageQueueChooser.getMessageQueues().size()) {
+				MessageQueue messageQueue;
+				synchronized (this.consumerMonitor) {
+					messageQueue = messageQueueChooser.choose();
+					messageQueueChooser.increment();
+				}
+
+				long offset = consumer.fetchConsumeOffset(messageQueue,
+						rocketMQConsumerProperties.getExtension().isFromStore());
+
+				log.debug("topic='{}', group='{}', messageQueue='{}', offset now='{}'",
+						this.topic, this.group, messageQueue, offset);
+
+				PullResult pullResult;
+				if (messageSelector != null) {
+					pullResult = consumer.pull(messageQueue, messageSelector, offset, 1);
+				}
+				else {
+					pullResult = consumer.pull(messageQueue, (String) null, offset, 1);
+				}
+
+				if (pullResult.getPullStatus() == PullStatus.FOUND) {
+					List<MessageExt> messageExtList = pullResult.getMsgFoundList();
+
+					Message message = RocketMQUtil
+							.convertToSpringMessage(messageExtList.get(0));
+
+					AcknowledgmentCallback ackCallback = this.ackCallbackFactory
+							.createCallback(new RocketMQAckInfo(messageQueue, pullResult,
+									consumer, offset));
+
+					Message messageResult = MessageBuilder.fromMessage(message).setHeader(
+							IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
+							ackCallback).build();
+					return messageResult;
+				}
+				else {
+					log.debug("messageQueue='{}' PullResult='{}' with topic `{}`",
+							messageQueueChooser.getMessageQueues(),
+							pullResult.getPullStatus(), topic);
+				}
+				count++;
+			}
+		}
+		catch (Exception e) {
+			log.error("Consumer pull error: " + e.getMessage(), e);
+		}
+		return null;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "rocketmq:message-source";
+	}
+
+	public synchronized void setRunning(boolean running) {
+		this.running = running;
+	}
+
+	public synchronized void resetMessageQueues(Set<MessageQueue> queueSet) {
+		log.info("resetMessageQueues, topic='{}', messageQueue=`{}`", topic, queueSet);
+		synchronized (this.consumerMonitor) {
+			this.messageQueueChooser.reset(queueSet);
+		}
+	}
+
+	public static class RocketMQCallbackFactory
+			implements AcknowledgmentCallbackFactory<RocketMQAckInfo> {
+
+		@Override
+		public AcknowledgmentCallback createCallback(RocketMQAckInfo info) {
+			return new RocketMQAckCallback(info);
+		}
+
+	}
+
+	public static class RocketMQAckCallback implements AcknowledgmentCallback {
+
+		private final RocketMQAckInfo ackInfo;
+
+		private boolean acknowledged;
+
+		private boolean autoAckEnabled = true;
+
+		public RocketMQAckCallback(RocketMQAckInfo ackInfo) {
+			this.ackInfo = ackInfo;
+		}
+
+		protected void setAcknowledged(boolean acknowledged) {
+			this.acknowledged = acknowledged;
+		}
+
+		@Override
+		public boolean isAcknowledged() {
+			return this.acknowledged;
+		}
+
+		@Override
+		public void noAutoAck() {
+			this.autoAckEnabled = false;
+		}
+
+		@Override
+		public boolean isAutoAck() {
+			return this.autoAckEnabled;
+		}
+
+		@Override
+		public void acknowledge(Status status) {
+			Assert.notNull(status, "'status' cannot be null");
+			if (this.acknowledged) {
+				throw new IllegalStateException("Already acknowledged");
+			}
+			log.debug("acknowledge(" + status.name() + ") for " + this);
+			synchronized (this.ackInfo.getConsumerMonitor()) {
+				try {
+					switch (status) {
+					case ACCEPT:
+					case REJECT:
+						ackInfo.getConsumer().updateConsumeOffset(
+								ackInfo.getMessageQueue(),
+								ackInfo.getPullResult().getNextBeginOffset());
+						log.debug("messageQueue='{}' offset update to `{}`",
+								ackInfo.getMessageQueue(), String.valueOf(
+										ackInfo.getPullResult().getNextBeginOffset()));
+						break;
+					case REQUEUE:
+						// decrease index and update offset of messageQueue of ackInfo
+						int oldIndex = ackInfo.getMessageQueueChooser().requeue();
+						ackInfo.getConsumer().updateConsumeOffset(
+								ackInfo.getMessageQueue(), ackInfo.getOldOffset());
+						log.debug(
+								"messageQueue='{}' offset requeue to index:`{}`, oldOffset:'{}'",
+								ackInfo.getMessageQueue(), oldIndex,
+								ackInfo.getOldOffset());
+						break;
+					default:
+						break;
+					}
+				}
+				catch (MQClientException e) {
+					log.error("acknowledge error: " + e.getErrorMessage(), e);
+				}
+				finally {
+					this.acknowledged = true;
+				}
+			}
+		}
+
+		@Override
+		public String toString() {
+			return "RocketMQAckCallback{" + "ackInfo=" + ackInfo + ", acknowledged="
+					+ acknowledged + ", autoAckEnabled=" + autoAckEnabled + '}';
+		}
+
+	}
+
+	public class RocketMQAckInfo {
+
+		private final MessageQueue messageQueue;
+
+		private final PullResult pullResult;
+
+		private final DefaultMQPullConsumer consumer;
+
+		private final long oldOffset;
+
+		public RocketMQAckInfo(MessageQueue messageQueue, PullResult pullResult,
+				DefaultMQPullConsumer consumer, long oldOffset) {
+			this.messageQueue = messageQueue;
+			this.pullResult = pullResult;
+			this.consumer = consumer;
+			this.oldOffset = oldOffset;
+		}
+
+		public MessageQueue getMessageQueue() {
+			return messageQueue;
+		}
+
+		public PullResult getPullResult() {
+			return pullResult;
+		}
+
+		public DefaultMQPullConsumer getConsumer() {
+			return consumer;
+		}
+
+		public RocketMQMessageQueueChooser getMessageQueueChooser() {
+			return RocketMQMessageSource.this.messageQueueChooser;
+		}
+
+		public long getOldOffset() {
+			return oldOffset;
+		}
+
+		public Object getConsumerMonitor() {
+			return RocketMQMessageSource.this.consumerMonitor;
+		}
+
+		@Override
+		public String toString() {
+			return "RocketMQAckInfo{" + "messageQueue=" + messageQueue + ", pullResult="
+					+ pullResult + ", consumer=" + consumer + ", oldOffset=" + oldOffset
+					+ '}';
+		}
+
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/metrics/Instrumentation.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.metrics;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class Instrumentation {
+
+	private final String name;
+
+	protected final AtomicBoolean started = new AtomicBoolean(false);
+
+	protected Exception startException = null;
+
+	public Instrumentation(String name) {
+		this.name = name;
+	}
+
+	public boolean isDown() {
+		return startException != null;
+	}
+
+	public boolean isUp() {
+		return started.get();
+	}
+
+	public boolean isOutOfService() {
+		return !started.get() && startException == null;
+	}
+
+	public void markStartedSuccessfully() {
+		started.set(true);
+	}
+
+	public void markStartFailed(Exception e) {
+		started.set(false);
+		startException = e;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public boolean isStarted() {
+		return started.get();
+	}
+
+	public Exception getStartException() {
+		return startException;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.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;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class InstrumentationManager {
+
+	private final Map<String, Object> runtime = new ConcurrentHashMap<>();
+
+	private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
+
+	public Set<Instrumentation> getHealthInstrumentations() {
+		return healthInstrumentations.entrySet().stream().map(Map.Entry::getValue)
+				.collect(Collectors.toSet());
+	}
+
+	public void addHealthInstrumentation(Instrumentation instrumentation) {
+		healthInstrumentations.put(instrumentation.getName(), instrumentation);
+	}
+
+	public Instrumentation getHealthInstrumentation(String key) {
+		return healthInstrumentations.get(key);
+	}
+
+	public Map<String, Object> getRuntime() {
+		return runtime;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.properties;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
+import org.apache.rocketmq.common.MixAll;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
+public class RocketMQBinderConfigurationProperties {
+
+	/**
+	 * The name server list for rocketMQ.
+	 */
+	private List<String> nameServer = Arrays
+			.asList(RocketMQBinderConstants.DEFAULT_NAME_SERVER);
+
+	/**
+	 * The property of "access-key".
+	 */
+	private String accessKey;
+
+	/**
+	 * The property of "secret-key".
+	 */
+	private String secretKey;
+
+	/**
+	 * Switch flag instance for message trace.
+	 */
+	private boolean enableMsgTrace = true;
+
+	/**
+	 * The name value of message trace topic.If you don't config,you can use the default
+	 * trace topic name.
+	 */
+	private String customizedTraceTopic = MixAll.RMQ_SYS_TRACE_TOPIC;
+
+	public List<String> getNameServer() {
+		return nameServer;
+	}
+
+	public void setNameServer(List<String> nameServer) {
+		this.nameServer = nameServer;
+	}
+
+	public String getAccessKey() {
+		return accessKey;
+	}
+
+	public void setAccessKey(String accessKey) {
+		this.accessKey = accessKey;
+	}
+
+	public String getSecretKey() {
+		return secretKey;
+	}
+
+	public void setSecretKey(String secretKey) {
+		this.secretKey = secretKey;
+	}
+
+	public boolean isEnableMsgTrace() {
+		return enableMsgTrace;
+	}
+
+	public void setEnableMsgTrace(boolean enableMsgTrace) {
+		this.enableMsgTrace = enableMsgTrace;
+	}
+
+	public String getCustomizedTraceTopic() {
+		return customizedTraceTopic;
+	}
+
+	public void setCustomizedTraceTopic(String customizedTraceTopic) {
+		this.customizedTraceTopic = customizedTraceTopic;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.properties;
+
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQBindingProperties implements BinderSpecificPropertiesProvider {
+
+	private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties();
+
+	private RocketMQProducerProperties producer = new RocketMQProducerProperties();
+
+	@Override
+	public RocketMQConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(RocketMQConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+
+	@Override
+	public RocketMQProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(RocketMQProducerProperties producer) {
+		this.producer = producer;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.properties;
+
+import java.util.Set;
+
+import com.alibaba.cloud.stream.binder.rocketmq.support.JacksonRocketMQHeaderMapper;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQConsumerProperties {
+
+	/**
+	 * using '||' to split tag {@link MQPushConsumer#subscribe(String, String)}.
+	 */
+	private String tags;
+
+	/**
+	 * {@link MQPushConsumer#subscribe(String, MessageSelector)}
+	 * {@link MessageSelector#bySql(String)}.
+	 */
+	private String sql;
+
+	/**
+	 * {@link MessageModel#BROADCASTING}.
+	 */
+	private Boolean broadcasting = false;
+
+	/**
+	 * if orderly is true, using {@link MessageListenerOrderly} else if orderly if false,
+	 * using {@link MessageListenerConcurrently}.
+	 */
+	private Boolean orderly = false;
+
+	/**
+	 * for concurrently listener. message consume retry strategy. see
+	 * {@link ConsumeConcurrentlyContext#delayLevelWhenNextConsume}. -1 means dlq(or
+	 * discard, see {@link this#shouldRequeue}), others means requeue.
+	 */
+	private int delayLevelWhenNextConsume = 0;
+
+	/**
+	 * for orderly listener. next retry delay time.
+	 */
+	private long suspendCurrentQueueTimeMillis = 1000;
+
+	private Boolean enabled = true;
+
+	/**
+	 * {@link JacksonRocketMQHeaderMapper#addTrustedPackages(String...)}.
+	 */
+	private Set<String> trustedPackages;
+
+	// ------------ For Pull Consumer ------------
+
+	private long pullTimeout = 10 * 1000;
+
+	private boolean fromStore;
+
+	// ------------ For Pull Consumer ------------
+
+	public String getTags() {
+		return tags;
+	}
+
+	public void setTags(String tags) {
+		this.tags = tags;
+	}
+
+	public String getSql() {
+		return sql;
+	}
+
+	public void setSql(String sql) {
+		this.sql = sql;
+	}
+
+	public Boolean getOrderly() {
+		return orderly;
+	}
+
+	public void setOrderly(Boolean orderly) {
+		this.orderly = orderly;
+	}
+
+	public Boolean getEnabled() {
+		return enabled;
+	}
+
+	public void setEnabled(Boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	public Boolean getBroadcasting() {
+		return broadcasting;
+	}
+
+	public void setBroadcasting(Boolean broadcasting) {
+		this.broadcasting = broadcasting;
+	}
+
+	public int getDelayLevelWhenNextConsume() {
+		return delayLevelWhenNextConsume;
+	}
+
+	public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
+		this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
+	}
+
+	public long getSuspendCurrentQueueTimeMillis() {
+		return suspendCurrentQueueTimeMillis;
+	}
+
+	public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
+		this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
+	}
+
+	public long getPullTimeout() {
+		return pullTimeout;
+	}
+
+	public void setPullTimeout(long pullTimeout) {
+		this.pullTimeout = pullTimeout;
+	}
+
+	public boolean isFromStore() {
+		return fromStore;
+	}
+
+	public void setFromStore(boolean fromStore) {
+		this.fromStore = fromStore;
+	}
+
+	public boolean shouldRequeue() {
+		return delayLevelWhenNextConsume != -1;
+	}
+
+	public Set<String> getTrustedPackages() {
+		return trustedPackages;
+	}
+
+	public void setTrustedPackages(Set<String> trustedPackages) {
+		this.trustedPackages = trustedPackages;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.properties;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@ConfigurationProperties("spring.cloud.stream.rocketmq")
+public class RocketMQExtendedBindingProperties extends
+		AbstractExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties, RocketMQBindingProperties> {
+
+	private static final String DEFAULTS_PREFIX = "spring.cloud.stream.rocketmq.default";
+
+	@Override
+	public String getDefaultsPrefix() {
+		return DEFAULTS_PREFIX;
+	}
+
+	@Override
+	public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
+		return RocketMQBindingProperties.class;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.properties;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQProducerProperties {
+
+	private Boolean enabled = true;
+
+	/**
+	 * Name of producer.
+	 */
+	private String group;
+
+	/**
+	 * Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}.
+	 */
+	private Integer maxMessageSize = 1024 * 1024 * 4;
+
+	private Boolean transactional = false;
+
+	private Boolean sync = false;
+
+	private Boolean vipChannelEnabled = true;
+
+	/**
+	 * Millis of send message timeout.
+	 */
+	private int sendMessageTimeout = 3000;
+
+	/**
+	 * Compress message body threshold, namely, message body larger than 4k will be
+	 * compressed on default.
+	 */
+	private int compressMessageBodyThreshold = 1024 * 4;
+
+	/**
+	 * Maximum number of retry to perform internally before claiming sending failure in
+	 * synchronous mode. This may potentially cause message duplication which is up to
+	 * application developers to resolve.
+	 */
+	private int retryTimesWhenSendFailed = 2;
+
+	/**
+	 * <p>
+	 * Maximum number of retry to perform internally before claiming sending failure in
+	 * asynchronous mode.
+	 * </p>
+	 * This may potentially cause message duplication which is up to application
+	 * developers to resolve.
+	 */
+	private int retryTimesWhenSendAsyncFailed = 2;
+
+	/**
+	 * Indicate whether to retry another broker on sending failure internally.
+	 */
+	private boolean retryNextServer = false;
+
+	public String getGroup() {
+		return group;
+	}
+
+	public void setGroup(String group) {
+		this.group = group;
+	}
+
+	public Boolean getEnabled() {
+		return enabled;
+	}
+
+	public void setEnabled(Boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	public Integer getMaxMessageSize() {
+		return maxMessageSize;
+	}
+
+	public void setMaxMessageSize(Integer maxMessageSize) {
+		this.maxMessageSize = maxMessageSize;
+	}
+
+	public Boolean getTransactional() {
+		return transactional;
+	}
+
+	public void setTransactional(Boolean transactional) {
+		this.transactional = transactional;
+	}
+
+	public Boolean getSync() {
+		return sync;
+	}
+
+	public void setSync(Boolean sync) {
+		this.sync = sync;
+	}
+
+	public Boolean getVipChannelEnabled() {
+		return vipChannelEnabled;
+	}
+
+	public void setVipChannelEnabled(Boolean vipChannelEnabled) {
+		this.vipChannelEnabled = vipChannelEnabled;
+	}
+
+	public int getSendMessageTimeout() {
+		return sendMessageTimeout;
+	}
+
+	public void setSendMessageTimeout(int sendMessageTimeout) {
+		this.sendMessageTimeout = sendMessageTimeout;
+	}
+
+	public int getCompressMessageBodyThreshold() {
+		return compressMessageBodyThreshold;
+	}
+
+	public void setCompressMessageBodyThreshold(int compressMessageBodyThreshold) {
+		this.compressMessageBodyThreshold = compressMessageBodyThreshold;
+	}
+
+	public int getRetryTimesWhenSendFailed() {
+		return retryTimesWhenSendFailed;
+	}
+
+	public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
+		this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
+	}
+
+	public int getRetryTimesWhenSendAsyncFailed() {
+		return retryTimesWhenSendAsyncFailed;
+	}
+
+	public void setRetryTimesWhenSendAsyncFailed(int retryTimesWhenSendAsyncFailed) {
+		this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
+	}
+
+	public boolean isRetryNextServer() {
+		return retryNextServer;
+	}
+
+	public void setRetryNextServer(boolean retryNextServer) {
+		this.retryNextServer = retryNextServer;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.provisioning;
+
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
+import org.apache.rocketmq.client.Validators;
+import org.apache.rocketmq.client.exception.MQClientException;
+
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+
+/**
+ * @author Timur Valiev
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQTopicProvisioner implements
+		ProvisioningProvider<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>> {
+
+	@Override
+	public ProducerDestination provisionProducerDestination(String name,
+			ExtendedProducerProperties<RocketMQProducerProperties> properties)
+			throws ProvisioningException {
+		checkTopic(name);
+		return new RocketProducerDestination(name);
+	}
+
+	@Override
+	public ConsumerDestination provisionConsumerDestination(String name, String group,
+			ExtendedConsumerProperties<RocketMQConsumerProperties> properties)
+			throws ProvisioningException {
+		checkTopic(name);
+		return new RocketConsumerDestination(name);
+	}
+
+	private void checkTopic(String topic) {
+		try {
+			Validators.checkTopic(topic);
+		}
+		catch (MQClientException e) {
+			throw new AssertionError(e);
+		}
+	}
+
+	private static final class RocketProducerDestination implements ProducerDestination {
+
+		private final String producerDestinationName;
+
+		RocketProducerDestination(String destinationName) {
+			this.producerDestinationName = destinationName;
+		}
+
+		@Override
+		public String getName() {
+			return producerDestinationName;
+		}
+
+		@Override
+		public String getNameForPartition(int partition) {
+			return producerDestinationName;
+		}
+
+	}
+
+	private static final class RocketConsumerDestination implements ConsumerDestination {
+
+		private final String consumerDestinationName;
+
+		RocketConsumerDestination(String consumerDestinationName) {
+			this.consumerDestinationName = consumerDestinationName;
+		}
+
+		@Override
+		public String getName() {
+			return this.consumerDestinationName;
+		}
+
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/provisioning/selector/PartitionMessageQueueSelector.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.provisioning.selector;
+
+import java.util.List;
+
+import org.apache.rocketmq.client.producer.MessageQueueSelector;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.stream.binder.BinderHeaders;
+
+/**
+ * @author wangxing
+ */
+public class PartitionMessageQueueSelector implements MessageQueueSelector {
+
+	private static final Logger LOGGER = LoggerFactory
+			.getLogger(PartitionMessageQueueSelector.class);
+
+	@Override
+	public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+		Integer partition = 0;
+		try {
+			partition = Math.abs(
+					Integer.parseInt(msg.getProperty(BinderHeaders.PARTITION_HEADER)));
+			if (partition >= mqs.size()) {
+				LOGGER.warn(
+						"the partition '{}' is greater than the number of queues '{}'.",
+						partition, mqs.size());
+				partition = partition % mqs.size();
+			}
+		}
+		catch (NumberFormatException ignored) {
+		}
+		return mqs.get(partition);
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/support/AbstractRocketMQHeaderMapper.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.support;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.rocketmq.common.message.MessageConst;
+
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.Assert;
+
+/**
+ * Base for RocketMQ header mappers.
+ *
+ * @author caotc
+ * @since 2.1.1.RELEASE
+ */
+public abstract class AbstractRocketMQHeaderMapper implements RocketMQHeaderMapper {
+
+	private static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
+
+	private Charset charset;
+
+	public AbstractRocketMQHeaderMapper() {
+		this(DEFAULT_CHARSET);
+	}
+
+	public AbstractRocketMQHeaderMapper(Charset charset) {
+		Assert.notNull(charset, "'charset' cannot be null");
+		this.charset = charset;
+	}
+
+	protected boolean matches(String headerName) {
+		return !MessageConst.STRING_HASH_SET.contains(headerName)
+				&& !MessageHeaders.ID.equals(headerName)
+				&& !MessageHeaders.TIMESTAMP.equals(headerName)
+				&& !MessageHeaders.CONTENT_TYPE.equals(headerName)
+				&& !MessageHeaders.REPLY_CHANNEL.equals(headerName)
+				&& !MessageHeaders.ERROR_CHANNEL.equals(headerName);
+	}
+
+	public Charset getCharset() {
+		return charset;
+	}
+
+	public void setCharset(Charset charset) {
+		Assert.notNull(charset, "'charset' cannot be null");
+		this.charset = charset;
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/support/JacksonRocketMQHeaderMapper.java

@@ -0,0 +1,287 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.support;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.ClassUtils;
+
+/**
+ * jackson header mapper for RocketMQ. Header types are added to a special header
+ * {@link #JSON_TYPES}.
+ *
+ * @author caotc
+ * @since 2.1.1.RELEASE
+ */
+public class JacksonRocketMQHeaderMapper extends AbstractRocketMQHeaderMapper {
+
+	private final static Logger log = LoggerFactory
+			.getLogger(JacksonRocketMQHeaderMapper.class);
+
+	private static final List<String> DEFAULT_TRUSTED_PACKAGES = Arrays
+			.asList("java.lang", "java.net", "java.util", "org.springframework.util");
+
+	/**
+	 * Header name for java types of other headers.
+	 */
+	public static final String JSON_TYPES = "spring_json_header_types";
+
+	private final ObjectMapper objectMapper;
+
+	private final Set<String> trustedPackages = new LinkedHashSet<>(
+			DEFAULT_TRUSTED_PACKAGES);
+
+	public JacksonRocketMQHeaderMapper(ObjectMapper objectMapper) {
+		this.objectMapper = objectMapper;
+	}
+
+	public JacksonRocketMQHeaderMapper(Charset charset, ObjectMapper objectMapper) {
+		super(charset);
+		this.objectMapper = objectMapper;
+	}
+
+	@Override
+	public Map<String, String> fromHeaders(MessageHeaders headers) {
+		final Map<String, String> target = new HashMap<>();
+		final Map<String, String> jsonHeaders = new HashMap<>();
+		headers.forEach((key, value) -> {
+			if (matches(key)) {
+				if (value instanceof String) {
+					target.put(key, (String) value);
+				}
+				else {
+					try {
+						String className = value.getClass().getName();
+						target.put(key, objectMapper.writeValueAsString(value));
+						jsonHeaders.put(key, className);
+					}
+					catch (Exception e) {
+						log.debug("Could not map " + key + " with type "
+								+ value.getClass().getName(), e);
+					}
+				}
+			}
+		});
+		if (jsonHeaders.size() > 0) {
+			try {
+				target.put(JSON_TYPES, objectMapper.writeValueAsString(jsonHeaders));
+			}
+			catch (IllegalStateException | JsonProcessingException e) {
+				log.error("Could not add json types header", e);
+			}
+		}
+		return target;
+	}
+
+	@Override
+	public MessageHeaders toHeaders(Map<String, String> source) {
+		final Map<String, Object> target = new HashMap<>();
+		final Map<String, String> jsonTypes = decodeJsonTypes(source);
+		source.forEach((key, value) -> {
+			if (matches(key) && !(key.equals(JSON_TYPES))) {
+				if (jsonTypes != null && jsonTypes.containsKey(key)) {
+					Class<?> type = Object.class;
+					String requestedType = jsonTypes.get(key);
+					boolean trusted = trusted(requestedType);
+					if (trusted) {
+						try {
+							type = ClassUtils.forName(requestedType, null);
+						}
+						catch (Exception e) {
+							log.error("Could not load class for header: " + key, e);
+						}
+					}
+
+					if (trusted) {
+						try {
+							Object val = decodeValue(value, type);
+							target.put(key, val);
+						}
+						catch (IOException e) {
+							log.error("Could not decode json type: " + value
+									+ " for key: " + key, e);
+							target.put(key, value);
+						}
+					}
+					else {
+						target.put(key, new NonTrustedHeaderType(value, requestedType));
+					}
+				}
+				else {
+					target.put(key, value);
+				}
+			}
+		});
+		return new MessageHeaders(target);
+	}
+
+	/**
+	 * @param packagesToTrust the packages to trust.
+	 * @see #addTrustedPackages(Collection)
+	 */
+	public void addTrustedPackages(String... packagesToTrust) {
+		if (Objects.nonNull(packagesToTrust)) {
+			addTrustedPackages(Arrays.asList(packagesToTrust));
+		}
+	}
+
+	/**
+	 * Add packages to the trusted packages list (default {@code java.util, java.lang})
+	 * used when constructing objects from JSON. If any of the supplied packages is
+	 * {@code "*"}, all packages are trusted. If a class for a non-trusted package is
+	 * encountered, the header is returned to the application with value of type
+	 * {@link NonTrustedHeaderType}.
+	 * @param packagesToTrust the packages to trust.
+	 */
+	public void addTrustedPackages(Collection<String> packagesToTrust) {
+		if (packagesToTrust != null) {
+			for (String whiteList : packagesToTrust) {
+				if ("*".equals(whiteList)) {
+					this.trustedPackages.clear();
+					break;
+				}
+				else {
+					this.trustedPackages.add(whiteList);
+				}
+			}
+		}
+	}
+
+	public Set<String> getTrustedPackages() {
+		return this.trustedPackages;
+	}
+
+	public ObjectMapper getObjectMapper() {
+		return objectMapper;
+	}
+
+	private Object decodeValue(String jsonString, Class<?> type)
+			throws IOException, LinkageError {
+		Object value = objectMapper.readValue(jsonString, type);
+		if (type.equals(NonTrustedHeaderType.class)) {
+			// Upstream NTHT propagated; may be trusted here...
+			NonTrustedHeaderType nth = (NonTrustedHeaderType) value;
+			if (trusted(nth.getUntrustedType())) {
+				try {
+					value = objectMapper.readValue(nth.getHeaderValue(),
+							ClassUtils.forName(nth.getUntrustedType(), null));
+				}
+				catch (Exception e) {
+					log.error("Could not decode header: " + nth, e);
+				}
+			}
+		}
+		return value;
+	}
+
+	@Nullable
+	private Map<String, String> decodeJsonTypes(Map<String, String> source) {
+		if (source.containsKey(JSON_TYPES)) {
+			String value = source.get(JSON_TYPES);
+			try {
+				return objectMapper.readValue(value,
+						new TypeReference<Map<String, String>>() {
+						});
+			}
+			catch (IOException e) {
+				log.error("Could not decode json types: " + value, e);
+			}
+		}
+		return null;
+	}
+
+	protected boolean trusted(String requestedType) {
+		if (requestedType.equals(NonTrustedHeaderType.class.getName())) {
+			return true;
+		}
+		if (!this.trustedPackages.isEmpty()) {
+			int lastDot = requestedType.lastIndexOf('.');
+			if (lastDot < 0) {
+				return false;
+			}
+			String packageName = requestedType.substring(0, lastDot);
+			for (String trustedPackage : this.trustedPackages) {
+				if (packageName.equals(trustedPackage)
+						|| packageName.startsWith(trustedPackage + ".")) {
+					return true;
+				}
+			}
+			return false;
+		}
+		return true;
+	}
+
+	/**
+	 * Represents a header that could not be decoded due to an untrusted type.
+	 */
+	public static class NonTrustedHeaderType {
+
+		private String headerValue;
+
+		private String untrustedType;
+
+		public NonTrustedHeaderType() {
+			super();
+		}
+
+		NonTrustedHeaderType(String headerValue, String untrustedType) {
+			this.headerValue = headerValue;
+			this.untrustedType = untrustedType;
+		}
+
+		public void setHeaderValue(String headerValue) {
+			this.headerValue = headerValue;
+		}
+
+		public String getHeaderValue() {
+			return this.headerValue;
+		}
+
+		public void setUntrustedType(String untrustedType) {
+			this.untrustedType = untrustedType;
+		}
+
+		public String getUntrustedType() {
+			return this.untrustedType;
+		}
+
+		@Override
+		public String toString() {
+			return "NonTrustedHeaderType [headerValue=" + headerValue + ", untrustedType="
+					+ this.untrustedType + "]";
+		}
+
+	}
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/support/RocketMQHeaderMapper.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq.support;
+
+import java.util.Map;
+
+import org.springframework.messaging.MessageHeaders;
+
+/**
+ * header value mapper for RocketMQ.
+ *
+ * @author caotc
+ * @since 2.1.1.RELEASE
+ */
+public interface RocketMQHeaderMapper {
+
+	/**
+	 * Map from the given {@link MessageHeaders} to the specified target message.
+	 * @param headers the abstracted MessageHeaders.
+	 * @return the native target message.
+	 */
+	Map<String, String> fromHeaders(MessageHeaders headers);
+
+	/**
+	 * Map from the given target message to abstracted {@link MessageHeaders}.
+	 * @param source the native target message.
+	 * @return the target headers.
+	 */
+	MessageHeaders toHeaders(Map<String, String> source);
+
+}

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/resources/META-INF/spring.binders

@@ -0,0 +1 @@
+rocketmq:com.alibaba.cloud.stream.binder.rocketmq.config.RocketMQBinderAutoConfiguration

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.alibaba.cloud.stream.binder.rocketmq.config.RocketMQComponent4BinderAutoConfiguration

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/test/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQAutoConfigurationTests.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.stream.binder.rocketmq;
+
+import java.util.Arrays;
+
+import com.alibaba.cloud.stream.binder.rocketmq.config.RocketMQBinderAutoConfiguration;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
+import org.junit.Test;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQAutoConfigurationTests {
+
+	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(
+					AutoConfigurations.of(RocketMQBinderAutoConfiguration.class))
+			.withPropertyValues(
+					"spring.cloud.stream.rocketmq.binder.name-server[0]=127.0.0.1:9876",
+					"spring.cloud.stream.rocketmq.binder.name-server[1]=127.0.0.1:9877",
+					"spring.cloud.stream.bindings.output.destination=TopicOrderTest",
+					"spring.cloud.stream.bindings.output.content-type=application/json",
+					"spring.cloud.stream.bindings.input1.destination=TopicOrderTest",
+					"spring.cloud.stream.bindings.input1.content-type=application/json",
+					"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=TopicOrderTest",
+					"spring.cloud.stream.bindings.input2.content-type=application/json",
+					"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=tag1");
+
+	@Test
+	public void testProperties() {
+		this.contextRunner.run(context -> {
+			RocketMQBinderConfigurationProperties binderConfigurationProperties = context
+					.getBean(RocketMQBinderConfigurationProperties.class);
+			assertThat(binderConfigurationProperties.getNameServer())
+					.isEqualTo(Arrays.asList("127.0.0.1:9876", "127.0.0.1:9877"));
+			RocketMQExtendedBindingProperties bindingProperties = context
+					.getBean(RocketMQExtendedBindingProperties.class);
+			assertThat(
+					bindingProperties.getExtendedConsumerProperties("input2").getTags())
+							.isEqualTo("tag1");
+			assertThat(bindingProperties.getExtendedConsumerProperties("input2")
+					.getOrderly()).isFalse();
+			assertThat(bindingProperties.getExtendedConsumerProperties("input1")
+					.getOrderly()).isTrue();
+		});
+	}
+
+}