@Override
	protected void doStart() {
		if (extendedConsumerProperties.getExtension() == null
				|| !extendedConsumerProperties.getExtension().getEnabled()) {
			return;
		}
		Instrumentation instrumentation = new Instrumentation(topic, this);
		try {
			pushConsumer.subscribe(topic, RocketMQUtils.getMessageSelector(
					extendedConsumerProperties.getExtension().getSubscription()));
			pushConsumer.start();
			instrumentation.markStartedSuccessfully();
		}
		catch (Exception e) {
			instrumentation.markStartFailed(e);
			log.error("DefaultMQPushConsumer init failed, Caused by " + e.getMessage());
			throw new MessagingException(MessageBuilder.withPayload(
					"DefaultMQPushConsumer init failed, Caused by " + e.getMessage())
					.build(), e);
		}
		finally {
			InstrumentationManager.addHealthInstrumentation(instrumentation);
		}
	}
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