diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
index a673aa96d..b21d5f13b 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
@@ -123,6 +123,18 @@ public class ConsumerProperties {
 	 */
 	private boolean useNativeDecoding;
 
+	/**
+	 * When set to true, the underlying binder will natively multiplex destinations on the same input binding.
+	 * For example, in the case of a comma separated multiple destinations, the core framework will skip binding
+	 * them individually if this is set to true, but delegate that responsibility to the binder.
+	 *
+	 * By default this property is set to `false` and the binder will individually bind each destinations in case
+	 * of a comma separated multi destination list. The individual binder implementations that need to support multiple
+	 * input bindings natively (multiplex) can enable this property. Under normal circumstances, the end users are
+	 * not expected to enable or disable this property directly.
+	 */
+	private boolean multiplex;
+
 	@Min(value = 1, message = "Concurrency should be greater than zero.")
 	public int getConcurrency() {
 		return concurrency;
@@ -209,4 +221,12 @@ public class ConsumerProperties {
 	public void setUseNativeDecoding(boolean useNativeDecoding) {
 		this.useNativeDecoding = useNativeDecoding;
 	}
+
+	public boolean isMultiplex() {
+		return multiplex;
+	}
+
+	public void setMultiplex(boolean multiplex) {
+		this.multiplex = multiplex;
+	}
 }
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
index 3b202f4a9..3633c74e5 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
@@ -54,6 +54,7 @@ import org.springframework.validation.beanvalidation.CustomValidatorBean;
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
  * @author Janne Valkealahti
+ * @author Soby Chacko
  */
 public class BindingService {
 
@@ -89,10 +90,6 @@ public class BindingService {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public <T> Collection<Binding<T>> bindConsumer(T input, String inputName) {
-		String bindingTarget = this.bindingServiceProperties
-				.getBindingDestination(inputName);
-		String[] bindingTargets = StringUtils
-				.commaDelimitedListToStringArray(bindingTarget);
 		Collection<Binding<T>> bindings = new ArrayList<>();
 		Binder<T, ConsumerProperties, ?> binder = (Binder<T, ConsumerProperties, ?>) getBinder(
 				inputName, input.getClass());
@@ -106,19 +103,30 @@ public class BindingService {
 			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
 			consumerProperties = extendedConsumerProperties;
 		}
+
 		validate(consumerProperties);
-		for (String target : bindingTargets) {
-			Binding<T> binding;
-			if (input instanceof PollableSource) {
-				binding = doBindPollableConsumer(input, inputName, binder, consumerProperties, target);
+
+		String bindingTarget = this.bindingServiceProperties
+				.getBindingDestination(inputName);
+
+		if (consumerProperties.isMultiplex()) {
+			bindings.add(doBindConsumer(input, inputName, binder, consumerProperties, bindingTarget));
+		}
+		else {
+			String[] bindingTargets = StringUtils
+					.commaDelimitedListToStringArray(bindingTarget);
+			for (String target : bindingTargets) {
+				Binding<T> binding;
+				if (input instanceof PollableSource) {
+					binding = doBindPollableConsumer(input, inputName, binder, consumerProperties, target);
+				} else {
+					binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
+				}
+				bindings.add(binding);
 			}
-			else {
-				binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
-			}
-			bindings.add(binding);
 		}
 		bindings = Collections.unmodifiableCollection(bindings);
-		this.consumerBindings.put(inputName, new ArrayList<Binding<?>>(bindings));
+		this.consumerBindings.put(inputName, new ArrayList<>(bindings));
 		return bindings;
 	}
 
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
index 15488dbe3..58d12b43d 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
@@ -81,6 +81,7 @@ import static org.mockito.Mockito.when;
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Janne Valkealahti
+ * @author Soby Chacko
  */
 public class BindingServiceTests {
 
@@ -163,6 +164,53 @@ public class BindingServiceTests {
 		binderFactory.destroy();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testConsumerBindingWhenMultiplexingIsEnabled() throws Exception {
+		BindingServiceProperties properties = new BindingServiceProperties();
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo,bar");
+
+		ConsumerProperties consumer = properties.getConsumerProperties("input");
+		consumer.setMultiplex(true);
+		props.setConsumer(consumer);
+
+		final String inputChannelName = "input";
+		bindingProperties.put(inputChannelName, props);
+
+		properties.setBindings(bindingProperties);
+
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
+
+		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
+		BindingService service = new BindingService(properties,
+				binderFactory);
+		MessageChannel inputChannel = new DirectChannel();
+
+		Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
+
+		when(binder.bindConsumer(eq("foo,bar"), isNull(), same(inputChannel),
+				any(ConsumerProperties.class))).thenReturn(mockBinding1);
+
+		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
+				"input");
+		assertThat(bindings).hasSize(1);
+
+		Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
+		Binding<MessageChannel> binding1 = iterator.next();
+
+		assertThat(binding1).isSameAs(mockBinding1);
+
+		service.unbindConsumers("input");
+
+		verify(binder).bindConsumer(eq("foo,bar"), isNull(), same(inputChannel),
+				any(ConsumerProperties.class));
+		verify(binding1).unbind();
+
+		binderFactory.destroy();
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testExplicitGroup() throws Exception {