diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
index 50424b03c..965be3954 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 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.
@@ -26,6 +26,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
@@ -129,24 +130,44 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 		Map<String, Binder> binders = this.context == null ? Collections.emptyMap()
 				: this.context.getBeansOfType(Binder.class);
 		Binder<T, ConsumerProperties, ProducerProperties> binder;
-		if (StringUtils.hasText(binderName) && binders.containsKey(binderName)) {
-			binder = (Binder<T, ConsumerProperties, ProducerProperties>) this.context
-					.getBean(binderName);
+
+		String bindingTargetTypeName = StringUtils.hasText(name) ? name : bindingTargetType.getSimpleName().toLowerCase();
+		final BinderConfiguration binderConfiguration = this.binderConfigurations.get(name);
+		if (binderConfiguration != null && binderConfiguration.getBinderType() != null) {
+			bindingTargetTypeName = binderConfiguration.getBinderType();
 		}
-		else if (binders.size() == 1) {
-			binder = binders.values().iterator().next();
-		}
-		else if (binders.size() > 1) {
-			throw new IllegalStateException(
-					"Multiple binders are available, however neither default nor "
-							+ "per-destination binder name is provided. Available binders are "
-							+ binders.keySet());
+
+		String[] bindingTargetTypeNameForLambda = new String[]{bindingTargetTypeName};
+		final Optional<String> foundBinder = binders.keySet().stream().filter(b -> b.equals(bindingTargetTypeNameForLambda[0])).findFirst();
+		if (foundBinder.isPresent()) { // found a match - now do all the customizations.
+			binder = binders.get(foundBinder.get());
+			if (!CollectionUtils.isEmpty(this.listeners)) {
+				for (Listener binderFactoryListener : this.listeners) {
+					binderFactoryListener.afterBinderContextInitialized(bindingTargetTypeName,
+						this.context);
+				}
+			}
 		}
 		else {
-			/*
-			 * This is the fall back to the old bootstrap that relies on spring.binders.
-			 */
-			binder = this.doGetBinder(binderName, bindingTargetType);
+			if (StringUtils.hasText(binderName) && binders.containsKey(binderName)) {
+				binder = (Binder<T, ConsumerProperties, ProducerProperties>) this.context
+					.getBean(binderName);
+			}
+			else if (binders.size() == 1) {
+				binder = binders.values().iterator().next();
+			}
+			else if (binders.size() > 1) {
+				throw new IllegalStateException(
+					"Multiple binders are available, however neither default nor "
+						+ "per-destination binder name is provided. Available binders are "
+						+ binders.keySet());
+			}
+			else {
+				/*
+				 * This is the fall back to the old bootstrap that relies on spring.binders.
+				 */
+				binder = this.doGetBinder(binderName, bindingTargetType);
+			}
 		}
 		if (this.binderCustomizer != null) {
 			this.binderCustomizer.customize(binder, binderName);