public class ConsumeMessageOrderlyService implements ConsumeMessageService {

    private static final InternalLogger log = ClientLogger.getLog();

    private final static long MAX_TIME_CONSUME_CONTINUOUSLY =

        Long.parseLong(System.getProperty("rocketmq.client.maxTimeConsumeContinuously", "60000"));//每次消费任务最大持续时间，默认60秒

    private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;//消息消费者实现类

    private final DefaultMQPushConsumer defaultMQPushConsumer;//消息消费者

    private final MessageListenerOrderly messageListener;//顺序消息消费监听器

    private final BlockingQueue<Runnable> consumeRequestQueue;//消息消费任务队列

    private final ThreadPoolExecutor consumeExecutor;//消息消费线程池

    private final String consumerGroup;//消息组名

    private final MessageQueueLock messageQueueLock = new MessageQueueLock();//消息消费端消息队列锁容器

    private final ScheduledExecutorService scheduledExecutorService;//调度任务线程池

    private volatile boolean stopped = false;

public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,

        MessageListenerOrderly messageListener) {

        this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;

        this.messageListener = messageListener;

        this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();

        this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();

        this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();

        this.consumeExecutor = new ThreadPoolExecutor(

            this.defaultMQPushConsumer.getConsumeThreadMin(),

            this.defaultMQPushConsumer.getConsumeThreadMax(),

            1000 * 60,

            TimeUnit.MILLISECONDS,

            this.consumeRequestQueue,

            new ThreadFactoryImpl("ConsumeMessageThread_"));

        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));

    }

public void start() {

        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {

            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override

                public void run() {

                    ConsumeMessageOrderlyService.this.lockMQPeriodically();

                }

            }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);

        }

    }

 if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())

                    || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
　　　　　　　　　　　　　　　　//消息消费逻辑
　　}

else {
    if (this.processQueue.isDropped()) {
        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
        return;
    }

    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
}