diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
index 4f49fee71e..13c1bbca54 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
@@ -17,6 +17,8 @@
 package org.springframework.integration.channel;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
@@ -30,6 +32,7 @@ import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 import reactor.util.context.ContextView;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.core.log.LogMessage;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.StaticMessageHeaderAccessor;
@@ -42,6 +45,9 @@ import org.springframework.util.Assert;
 /**
  * The {@link AbstractMessageChannel} implementation for the
  * Reactive Streams {@link Publisher} based on the Project Reactor {@link Flux}.
+ * <p>
+ * This class implements {@link Lifecycle} to control subscriptions to publishers
+ * attached via {@link #subscribeTo(Publisher)}, when this channel is restarted.
  *
  * @author Artem Bilan
  * @author Gary Russell
@@ -50,11 +56,13 @@ import org.springframework.util.Assert;
  * @since 5.0
  */
 public class FluxMessageChannel extends AbstractMessageChannel
-		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
+		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel, Lifecycle {
 
 	private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
 
-	private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
+	private final List<Publisher<? extends Message<?>>> sourcePublishers = new ArrayList<>();
+
+	private volatile Disposable.Composite upstreamSubscriptions = Disposables.composite();
 
 	private volatile boolean active = true;
 
@@ -111,6 +119,56 @@ public class FluxMessageChannel extends AbstractMessageChannel
 				.subscribe(subscriber);
 	}
 
+	@Override
+	public void start() {
+		this.active = true;
+		this.upstreamSubscriptions = Disposables.composite();
+		this.sourcePublishers.forEach(this::doSubscribeTo);
+	}
+
+	@Override
+	public void stop() {
+		this.active = false;
+		this.upstreamSubscriptions.dispose();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.active;
+	}
+
+	private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableReference) {
+		Disposable disposable = disposableReference.get();
+		if (disposable != null) {
+			this.upstreamSubscriptions.remove(disposable);
+			disposable.dispose();
+		}
+	}
+
+	@Override
+	public void subscribeTo(Publisher<? extends Message<?>> publisher) {
+		this.sourcePublishers.add(publisher);
+		doSubscribeTo(publisher);
+	}
+
+	private void doSubscribeTo(Publisher<? extends Message<?>> publisher) {
+		Flux<Object> upstreamPublisher =
+				Flux.from(publisher)
+						.doOnComplete(() -> this.sourcePublishers.remove(publisher))
+						.delaySubscription(
+								Mono.fromCallable(this.sink::currentSubscriberCount)
+										.filter((value) -> value > 0)
+										.repeatWhenEmpty((repeat) ->
+												this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat))
+						.flatMap((message) ->
+								Mono.just(message)
+										.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
+										.contextWrite(StaticMessageHeaderAccessor.getReactorContext(message)))
+						.contextCapture();
+
+		addPublisherToSubscribe(upstreamPublisher);
+	}
+
 	private void addPublisherToSubscribe(Flux<?> publisher) {
 		AtomicReference<Disposable> disposableReference = new AtomicReference<>();
 
@@ -126,32 +184,6 @@ public class FluxMessageChannel extends AbstractMessageChannel
 		}
 	}
 
-	private void disposeUpstreamSubscription(AtomicReference<Disposable> disposableReference) {
-		Disposable disposable = disposableReference.get();
-		if (disposable != null) {
-			this.upstreamSubscriptions.remove(disposable);
-			disposable.dispose();
-		}
-	}
-
-	@Override
-	public void subscribeTo(Publisher<? extends Message<?>> publisher) {
-		Flux<Object> upstreamPublisher =
-				Flux.from(publisher)
-						.delaySubscription(
-								Mono.fromCallable(this.sink::currentSubscriberCount)
-										.filter((value) -> value > 0)
-										.repeatWhenEmpty((repeat) ->
-												this.active ? repeat.delayElements(Duration.ofMillis(100)) : repeat))
-						.flatMap((message) ->
-								Mono.just(message)
-										.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
-										.contextWrite(StaticMessageHeaderAccessor.getReactorContext(message)))
-						.contextCapture();
-
-		addPublisherToSubscribe(upstreamPublisher);
-	}
-
 	private void sendReactiveMessage(Message<?> message) {
 		Message<?> messageToSend = message;
 		// We have just restored Reactor context, so no need in a header anymore.
@@ -169,7 +201,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 			}
 		}
 		catch (Exception ex) {
-			logger.warn(ex, LogMessage.format("Error during processing event: %s", messageToSend));
+			logger.error(ex, LogMessage.format("Error during processing event: %s", messageToSend));
 		}
 	}
 
@@ -177,6 +209,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 	public void destroy() {
 		this.active = false;
 		this.upstreamSubscriptions.dispose();
+		this.sourcePublishers.clear();
 		this.sink.emitComplete(Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1)));
 		super.destroy();
 	}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
index 24cd97399b..b1a4094b6f 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
@@ -38,6 +38,7 @@ import reactor.test.StepVerifier;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -266,6 +267,35 @@ public class ReactiveStreamsTests {
 				.verify(Duration.ofSeconds(10));
 	}
 
+	@Autowired
+	QueueChannel fromPublisherResult;
+
+	@Autowired
+	ConfigurableApplicationContext applicationContext;
+
+	@Test
+	@DirtiesContext
+		// Use disruptive this.applicationContext.start()
+	void verifyFluxMessageChannelRestart() {
+		for (long i = 0; i < 3L; i++) {
+			assertThat(this.fromPublisherResult.receive(10_000)).extracting(Message::getPayload).isEqualTo(i);
+		}
+
+		this.applicationContext.stop();
+
+		this.fromPublisherResult.purge(null);
+
+		this.applicationContext.start();
+
+		// The applicationContext restart causes all the endpoint to be started,
+		// while we really don't have a subscription to this producer
+		this.testMessageProducer.stop();
+
+		for (long i = 0; i < 3L; i++) {
+			assertThat(this.fromPublisherResult.receive(10_000)).extracting(Message::getPayload).isEqualTo(i);
+		}
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -325,6 +355,13 @@ public class ReactiveStreamsTests {
 					.toReactivePublisher(true);
 		}
 
+		@Bean
+		IntegrationFlow fromPublisher() {
+			return IntegrationFlow.from(Flux.interval(Duration.ofMillis(100)).map(GenericMessage::new))
+					.channel(c -> c.queue("fromPublisherResult"))
+					.get();
+		}
+
 	}
 
 	private static class TestMessageProducerSpec