diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
index f19269806c..3f274857b8 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
@@ -30,6 +30,8 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.support.management.PollableChannelManagement;
+import org.springframework.integration.support.management.metrics.CounterFacade;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -54,6 +56,8 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	private Queue queue;
 
+	private CounterFacade receiveCounter;
+
 	private volatile int executorInterceptorsSize;
 
 	private volatile boolean declared;
@@ -161,11 +165,13 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 	}
 
 	@Override
+	@Nullable
 	public Message<?> receive() {
 		return doReceive(null);
 	}
 
 	@Override
+	@Nullable
 	public Message<?> receive(long timeout) {
 		return doReceive(timeout);
 	}
@@ -196,6 +202,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 			}
 			else {
 				if (countsEnabled) {
+					if (getMetricsCaptor() != null) {
+						incrementReceiveCounter();
+					}
 					getMetrics().afterReceive();
 					counted = true;
 				}
@@ -211,6 +220,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 					logger.debug("postReceive on channel '" + this + "', message: " + message);
 				}
 			}
+
 			if (interceptorStack != null) {
 				if (message != null) {
 					message = interceptorList.postReceive(message, this);
@@ -221,6 +231,16 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 		}
 		catch (RuntimeException e) {
 			if (countsEnabled && !counted) {
+				if (getMetricsCaptor() != null) {
+					getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+							.tag("type", "channel")
+							.tag("result", "failure")
+							.tag("exception", e.getClass().getSimpleName())
+							.description("Messages received")
+							.build()
+							.increment();
+				}
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {
@@ -269,6 +289,20 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 		}
 	}
 
+	private void incrementReceiveCounter() {
+		if (this.receiveCounter == null) {
+			this.receiveCounter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+					.tag("name", getComponentName())
+					.tag("type", "channel")
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Messages received")
+					.build();
+		}
+		this.receiveCounter.increment();
+	}
+
+
 	@Override
 	public void setInterceptors(List<ChannelInterceptor> interceptors) {
 		super.setInterceptors(interceptors);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
index 51edfe3663..f6a5a9a406 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
@@ -27,7 +27,6 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ExecutorChannelInterceptor;
-import org.springframework.util.CollectionUtils;
 
 /**
  * Base class for all pollable channels.
@@ -98,7 +97,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 		boolean counted = false;
 		boolean countsEnabled = isCountsEnabled();
 		try {
-			if (logger.isTraceEnabled()) {
+			if (isLoggingEnabled() && logger.isTraceEnabled()) {
 				logger.trace("preReceive on channel '" + this + "'");
 			}
 			if (interceptorList.getSize() > 0) {
@@ -109,20 +108,27 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 				}
 			}
 			Message<?> message = doReceive(timeout);
-			if (countsEnabled && message != null) {
-				if (getMetricsCaptor() != null) {
-					incrementReceiveCounter();
+			if (message == null) {
+				if (isLoggingEnabled() && logger.isTraceEnabled()) {
+					logger.trace("postReceive on channel '" + this + "', message is null");
 				}
-				getMetrics().afterReceive();
-				counted = true;
 			}
-			if (message != null && logger.isDebugEnabled()) {
-				logger.debug("postReceive on channel '" + this + "', message: " + message);
+			else {
+				if (countsEnabled) {
+					if (getMetricsCaptor() != null) {
+						incrementReceiveCounter();
+					}
+					getMetrics().afterReceive();
+					counted = true;
+				}
+
+				if (isLoggingEnabled() && logger.isDebugEnabled()) {
+						logger.debug("postReceive on channel '" + this + "', message: " + message);
+				}
+
 			}
-			else if (logger.isTraceEnabled()) {
-				logger.trace("postReceive on channel '" + this + "', message is null");
-			}
-			if (!CollectionUtils.isEmpty(interceptorStack)) {
+
+			if (interceptorStack != null) {
 				if (message != null) {
 					message = interceptorList.postReceive(message, this);
 				}
@@ -144,7 +150,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 				}
 				getMetrics().afterError();
 			}
-			if (!CollectionUtils.isEmpty(interceptorStack)) {
+			if (interceptorStack != null) {
 				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
 			}
 			throw e;
@@ -219,10 +225,10 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 	 * return immediately with or without success). A negative timeout value
 	 * indicates that the method should block until either a message is
 	 * available or the blocking thread is interrupted.
-	 *
 	 * @param timeout The timeout.
 	 * @return The message, or null.
 	 */
+	@Nullable
 	protected abstract Message<?> doReceive(long timeout);
 
 }
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
index 3c50936c81..c8dd281e03 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
@@ -22,7 +22,9 @@ import java.util.List;
 
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.support.management.PollableChannelManagement;
+import org.springframework.integration.support.management.metrics.CounterFacade;
 import org.springframework.jms.core.JmsTemplate;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -41,6 +43,8 @@ public class PollableJmsChannel extends AbstractJmsChannel
 
 	private String messageSelector;
 
+	private CounterFacade receiveCounter;
+
 	private volatile int executorInterceptorsSize;
 
 	public PollableJmsChannel(JmsTemplate jmsTemplate) {
@@ -72,17 +76,30 @@ public class PollableJmsChannel extends AbstractJmsChannel
 	}
 
 	@Override
+	@Nullable
+	public Message<?> receive(long timeout) {
+		try {
+			DynamicJmsTemplateProperties.setReceiveTimeout(timeout);
+			return receive();
+		}
+		finally {
+			DynamicJmsTemplateProperties.clearReceiveTimeout();
+		}
+	}
+
+	@Override
+	@Nullable
 	public Message<?> receive() {
 		ChannelInterceptorList interceptorList = getInterceptors();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;
 		boolean countsEnabled = isCountsEnabled();
 		try {
-			if (logger.isTraceEnabled()) {
+			if (isLoggingEnabled() && logger.isTraceEnabled()) {
 				logger.trace("preReceive on channel '" + this + "'");
 			}
 			if (interceptorList.getInterceptors().size() > 0) {
-				interceptorStack = new ArrayDeque<ChannelInterceptor>();
+				interceptorStack = new ArrayDeque<>();
 
 				if (!interceptorList.preReceive(this, interceptorStack)) {
 					return null;
@@ -98,12 +115,15 @@ public class PollableJmsChannel extends AbstractJmsChannel
 
 			Message<?> message = null;
 			if (object == null) {
-				if (logger.isTraceEnabled()) {
+				if (isLoggingEnabled() && logger.isTraceEnabled()) {
 					logger.trace("postReceive on channel '" + this + "', message is null");
 				}
 			}
 			else {
 				if (countsEnabled) {
+					if (getMetricsCaptor() != null) {
+						incrementReceiveCounter();
+					}
 					getMetrics().afterReceive();
 					counted = true;
 				}
@@ -115,7 +135,7 @@ public class PollableJmsChannel extends AbstractJmsChannel
 							.withPayload(object)
 							.build();
 				}
-				if (logger.isDebugEnabled()) {
+				if (isLoggingEnabled() && logger.isDebugEnabled()) {
 					logger.debug("postReceive on channel '" + this + "', message: " + message);
 				}
 			}
@@ -129,6 +149,16 @@ public class PollableJmsChannel extends AbstractJmsChannel
 		}
 		catch (RuntimeException e) {
 			if (countsEnabled && !counted) {
+				if (getMetricsCaptor() != null) {
+					getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+							.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+							.tag("type", "channel")
+							.tag("result", "failure")
+							.tag("exception", e.getClass().getSimpleName())
+							.description("Messages received")
+							.build()
+							.increment();
+				}
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {
@@ -138,15 +168,17 @@ public class PollableJmsChannel extends AbstractJmsChannel
 		}
 	}
 
-	@Override
-	public Message<?> receive(long timeout) {
-		try {
-			DynamicJmsTemplateProperties.setReceiveTimeout(timeout);
-			return receive();
-		}
-		finally {
-			DynamicJmsTemplateProperties.clearReceiveTimeout();
+	private void incrementReceiveCounter() {
+		if (this.receiveCounter == null) {
+			this.receiveCounter = getMetricsCaptor().counterBuilder(RECEIVE_COUNTER_NAME)
+					.tag("name", getComponentName())
+					.tag("type", "channel")
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Messages received")
+					.build();
 		}
+		this.receiveCounter.increment();
 	}
 
 	@Override