diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
index e47a0eb..d50515b 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
@@ -16,9 +16,16 @@
 
 package org.springframework.integration.dsl;
 
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 
 import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.context.Lifecycle;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.expression.MethodFilter;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
@@ -44,6 +51,7 @@ import org.springframework.integration.filter.MethodInvokingSelector;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.handler.DelayHandler;
+import org.springframework.integration.handler.ExpressionCommandMessageProcessor;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
@@ -57,9 +65,13 @@ import org.springframework.integration.transformer.HeaderFilter;
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
 import org.springframework.integration.transformer.Transformer;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
+import org.springframework.util.CustomizableThreadCreator;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -114,6 +126,10 @@ public final class IntegrationFlowBuilder {
 		return this.channel(messageChannelSpec.get());
 	}
 
+	public IntegrationFlowBuilder controlBus() {
+		return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(new ControlBusMethodFilter())), null);
+	}
+
 	public IntegrationFlowBuilder transform(String expression) {
 		Assert.hasText(expression);
 		return this.transform(new ExpressionEvaluatingTransformer(PARSER.parseExpression(expression)));
@@ -438,4 +454,40 @@ public final class IntegrationFlowBuilder {
 		return this.flow;
 	}
 
+
+	private static class ControlBusMethodFilter implements MethodFilter {
+
+		public List<Method> filter(List<Method> methods) {
+			List<Method> supportedMethods = new ArrayList<Method>();
+			for (Method method : methods) {
+				if (this.accept(method)) {
+					supportedMethods.add(method);
+				}
+			}
+			return supportedMethods;
+		}
+
+		private boolean accept(Method method) {
+			Class<?> declaringClass = method.getDeclaringClass();
+			if (Lifecycle.class.isAssignableFrom(declaringClass)
+					&& ReflectionUtils.findMethod(Lifecycle.class, method.getName(), method.getParameterTypes()) != null) {
+				return true;
+			}
+			if (CustomizableThreadCreator.class.isAssignableFrom(declaringClass)
+					&& (method.getName().startsWith("get")
+					|| method.getName().startsWith("set")
+					|| method.getName().startsWith("shutdown"))) {
+				return true;
+			}
+			if (this.hasAnnotation(method, ManagedAttribute.class) || this.hasAnnotation(method, ManagedOperation.class)) {
+				return true;
+			}
+			return false;
+		}
+
+		private boolean hasAnnotation(Method method, Class<? extends Annotation> annotationType) {
+			return AnnotationUtils.findAnnotation(method, annotationType) != null;
+		}
+	}
+
 }
diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
index af22a9e..1cce053 100644
--- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
+++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
@@ -50,7 +50,6 @@ import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationListener;
-import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -110,6 +109,10 @@ public class IntegrationFlowTests {
 	@Autowired
 	private ListableBeanFactory beanFactory;
 
+	@Autowired
+	@Qualifier("controlBus")
+	private MessageChannel controlBus;
+
 	@Autowired
 	@Qualifier("flow1QueueChannel")
 	private PollableChannel outputChannel;
@@ -207,7 +210,7 @@ public class IntegrationFlowTests {
 			assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
 			assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
 		}
-		this.beanFactory.getBean("payloadSerializingTransformer", Lifecycle.class).start();
+		this.controlBus.send(new GenericMessage<Object>("@payloadSerializingTransformer.start()"));
 
 		final AtomicBoolean used = new AtomicBoolean();
 
@@ -253,7 +256,7 @@ public class IntegrationFlowTests {
 			assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
 			assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
 		}
-		this.beanFactory.getBean("bridge", Lifecycle.class).start();
+		this.controlBus.send(new GenericMessage<Object>("@bridge.start()"));
 		this.bridgeFlow2Input.send(message);
 		reply = this.bridgeFlow2Output.receive(5000);
 		assertNotNull(reply);
@@ -396,7 +399,7 @@ public class IntegrationFlowTests {
 			assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
 		}
 
-		this.beanFactory.getBean("xpathHeaderEnricher", Lifecycle.class).start();
+		this.controlBus.send(new GenericMessage<Object>("@xpathHeaderEnricher.start()"));
 		this.xpathHeaderEnricherInput.send(message);
 
 		Message<?> result = replyChannel.receive(2000);
@@ -418,6 +421,11 @@ public class IntegrationFlowTests {
 			return source;
 		}
 
+		@Bean
+		public IntegrationFlow controlBusFlow() {
+			return IntegrationFlows.from("controlBus").controlBus().get();
+		}
+
 		@Bean
 		public IntegrationFlow flow1() {
 			return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100)))