diff --git a/core/spring-cloud-stream/pom.xml b/core/spring-cloud-stream/pom.xml
index c573abe70..3d9c6b2e6 100644
--- a/core/spring-cloud-stream/pom.xml
+++ b/core/spring-cloud-stream/pom.xml
@@ -65,6 +65,12 @@
 			<artifactId>spring-core-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+		    <groupId>io.micrometer</groupId>
+		    <artifactId>context-propagation</artifactId>
+		    <version>1.1.0</version>
+		    <scope>provided</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-autoconfigure-processor</artifactId>
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
index 3f9737ecf..1d470468a 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
@@ -21,8 +21,13 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import io.micrometer.context.ContextExecutorService;
+import io.micrometer.context.ContextSnapshotFactory;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -47,6 +52,8 @@ import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.core.ResolvableType;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.integration.channel.ExecutorChannel;
 import org.springframework.integration.config.GlobalChannelInterceptorProcessor;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.lang.Nullable;
@@ -54,11 +61,13 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
+
 /**
  * A class which allows user to send data to an output binding.
  * While in a common scenario of a typical spring-cloud-stream application user rarely
@@ -96,12 +105,19 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 
 	private boolean initialized;
 
+	private boolean async;
+
 	private final BindingService bindingService;
 
 	private final Map<Integer, FunctionInvocationWrapper> streamBridgeFunctionCache;
 
 	private final FunctionInvocationHelper<?> functionInvocationHelper;
 
+	private ExecutorService executorService;
+	
+	private static final boolean isContextPropagationPresent = ClassUtils.isPresent(
+			"io.micrometer.context.ContextSnapshotFactory", StreamBridge.class.getClassLoader());
+
 	/**
 	 *
 	 * @param functionCatalog instance of {@link FunctionCatalog}
@@ -111,6 +127,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 	@SuppressWarnings("serial")
 	StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties,
 		ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) {
+		this.executorService = Executors.newCachedThreadPool();
 		Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
 		Assert.notNull(applicationContext, "'applicationContext' must not be null");
 		Assert.notNull(bindingServiceProperties, "'bindingServiceProperties' must not be null");
@@ -253,9 +270,9 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 				}
 			}
 			else {
-				messageChannel = new DirectWithAttributesChannel();
-				((DirectWithAttributesChannel) messageChannel).setApplicationContext(applicationContext);
-				((DirectWithAttributesChannel) messageChannel).setComponentName(destinationName);
+				messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
+				((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
+				((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
 				if (this.destinationBindingCallback != null) {
 					Object extendedProducerProperties = this.bindingService
 							.getExtendedProducerProperties(messageChannel, destinationName);
@@ -310,8 +327,36 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 
 	@Override
 	public void destroy() throws Exception {
+		if (!this.executorService.awaitTermination(10000, TimeUnit.MILLISECONDS)) {
+			logger.warn("Failed to terminate executor. Terminating current tasks.");
+			this.executorService.shutdownNow();
+		}
+		else {
+			this.executorService.shutdown();
+		}
+		
+		this.executorService = null;
+		this.async = false;
 		channelCache.keySet().forEach(bindingService::unbindProducers);
 		channelCache.clear();
 	}
 
+	public boolean isAsync() {
+		return async;
+	}
+
+	public void setAsync(boolean async) {
+		if (isContextPropagationPresent) {
+			this.executorService = ContextPropagationHelper.wrap(this.executorService);
+		}
+		this.executorService = ContextExecutorService
+				.wrap(Executors.newCachedThreadPool(), () -> ContextSnapshotFactory.builder().build().captureAll());
+		this.async = async;
+	}
+	
+	private static final class ContextPropagationHelper {
+		static ExecutorService wrap(ExecutorService executorService) {
+			return ContextExecutorService.wrap(executorService, () -> ContextSnapshotFactory.builder().build().captureAll());
+		}
+	}
 }
diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc
index 78271c98a..d919f0aa8 100644
--- a/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc
+++ b/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc
@@ -293,6 +293,24 @@ Also, note that `streamBridge.send(..)` method takes an `Object` for data. This
 will go through the same routine when sending output as if it was from any Function or Supplier providing the same level
 of consistency as with functions. This means the output type conversion, partitioning etc are honored as if it was from the output produced by functions.
 
+
+==== StreamBridge with async send
+
+`StreamBridge` uses sending mechanism provided by _Spring Integration_ framework which is at the core of the _Spring Cloud Stream_. By default this mechanism uses the sender’s thread. In other words, the send is blocking. While this is ok for many cases, there are cases when you want such send to be async. To do that use `setAsync(true)` method of the `StreamBridge` before invoking one of the send methods.
+
+**Observability Context propagation with asynchronous send**
+
+When using Observability support provided by the framework as well as supporting Spring frameworks, breaking thread boundaries will affect consistency of Observability context, thus your tracing history.  To avoid that all you need is to add `context-propagation` dependency form Micrometer (see below)
+
+[source, xml]
+----
+<dependency>
+    <groupId>io.micrometer</groupId>
+    <artifactId>context-propagation</artifactId>
+    <version>1.1.0</version>
+</dependency>
+----
+
 [[streambridge-and-dynamic-destinations]]
 ==== StreamBridge and Dynamic Destinations