private <R> R consumeMessage(List<MessageExt> messageExtList,
			Supplier<R> failSupplier, Supplier<R> sucSupplier) {
		if (CollectionUtils.isEmpty(messageExtList)) {
			throw new MessagingException(
					"DefaultMQPushConsumer consuming failed, Caused by messageExtList is empty");
		}
		for (MessageExt messageExt : messageExtList) {
			try {
				Message<?> message = RocketMQMessageConverterSupport
						.convertMessage2Spring(messageExt);
				if (this.retryTemplate != null) {
					this.retryTemplate.execute(context -> {
						this.sendMessage(message);
						return message;
					}, this.recoveryCallback);
				}
				else {
					this.sendMessage(message);
				}
			}
			catch (Exception e) {
				log.warn("consume message failed. messageExt:{}", messageExt, e);
				return failSupplier.get();
			}
		}
		return sucSupplier.get();
	}
The actual execution of a user-defined input consumption service method.
@param messageExtList rocket mq message list
@param failSupplier {@link ConsumeConcurrentlyStatus} or
    {@link ConsumeOrderlyStatus}
@param sucSupplier {@link ConsumeConcurrentlyStatus} or
    {@link ConsumeOrderlyStatus}
@param <R> object
@return R
spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/RocketMQInboundChannelAdapter.java