@Override
	protected synchronized Object doReceive() {
		if (messageExtIterator == null) {
			List<MessageExt> messageExtList = consumer.poll();
			if (CollectionUtils.isEmpty(messageExtList)) {
				return null;
			}
			messageExtIterator = messageExtList.iterator();
		}
		MessageExt messageExt = messageExtIterator.next();
		if (!messageExtIterator.hasNext()) {
			messageExtIterator = null;
		}
		if (null == messageExt) {
			return null;
		}
		MessageQueue messageQueue = this.acquireCurrentMessageQueue(messageExt.getTopic(),
				messageExt.getQueueId(), messageExt.getBrokerName());
		if (messageQueue == null) {
			throw new IllegalArgumentException(
					"The message queue is not in assigned list");
		}
		Message message = RocketMQMessageConverterSupport
				.convertMessage2Spring(messageExt);
		return MessageBuilder.fromMessage(message)
				.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
						new RocketMQAckCallback(this.consumer, messageQueue, messageExt))
				.build();
	}
@author <a href="mailto:fangjian0423@gmail.com">Jim</a>
spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/pull/RocketMQMessageSource.java