12 结合实际场景再聊 DefaultLitePullConsumer 的使用

通过上文的讲解，各位读者朋友们应该对 DefaultLitePullConsumer 有了一个全面的理解，但会不会觉得意犹未尽之感，因为在实战环节只是给出了一个 Demo 级别的示例，本篇将一个大数据领域的消息拉取批处理场景丰富一些 DefaultLitePullConsumer 的使用技巧。

场景描述

现在订单系统会将消息发送到 ORDER_TOPIC 中，大数据这边需要将订单数据导入自己的计算平台，对用户、商家的订单行为进行分析。

PUSH 与 PULL 模式选型

大数据这边只需订阅 ORDER_TOPIC 主题就可以完成数据的同步，那是采用 PUSH 模式还是 PULL 模式呢？

大数据领域通常采用 PULL 模式，因为大数据数据计算都是基于 Spark 等批处理框架，基本都是批处理任务，例如每 5 分钟、每 10 分钟执行一次，而且一个批次能处理的数据越多越好，这样有利于大量数据分布式计算，整体性能计算效能更佳，如果采用 PUSH 模式，虽然也可以指定一次拉取的消息调试，但由于 PUSH 模式是几乎实时的，故每次拉取时服务端几乎不可能挤满了大量的消息，导致一次拉取的消息其实不多，再者是对于一个消费 JVM 来说，面对一个 RocketMQ 集群只会开启一条线程进行消息拉取，而 PULL 模式每一个消费者就可以指定多个消息拉取线程（默认为 20 个），故从消息拉取效能这个方面，PULL 模式占优，并且这个对实时性要求没那么高，故 综合考虑下来，该场景最终采用 PULL 模式。

方案设计

大概的实现思路如下图所示：

代码实现与代码解读

// BigDataPullConsumer.java
package org.apache.rocketmq.example.simple.litepull;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class BigDataPullConsumer {

    private final ExecutorService executorService = new ThreadPoolExecutor(30, 30, 0L,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000), new DefaultThreadFactory("business-executer-
                                                                                        "));

    private final ExecutorService pullTaskExecutor = new ThreadPoolExecutor(1, 1, 0L,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new DefaultThreadFactory("pull-batch-"));

    private String consumerGroup;
    private String nameserverAddr;
    private String topic;
    private String filter;
    private MessageListener messageListener;
    private DefaultMQProducer rertyMQProducer;
    private PullBatchTask pullBatchTask;

    public BigDataPullConsumer(String consumerGroup, String nameserverAddr, String topic, String filter) {
        this.consumerGroup = consumerGroup;
        this.nameserverAddr = nameserverAddr;
        this.topic = topic;
        this.filter = filter;
        initRetryMQProducer();
    }

    private void initRetryMQProducer() {
        this.rertyMQProducer = new DefaultMQProducer(consumerGroup + "-retry");
        this.rertyMQProducer.setNamesrvAddr(this.nameserverAddr);
        try {
            this.rertyMQProducer.start();
        } catch (Throwable e) {
            throw new RuntimeException("启动失败", e);
        }

    }

    public void registerMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public void start() {
        //没有考虑重复调用问题
        this.pullBatchTask = new PullBatchTask(consumerGroup, nameserverAddr, topic,filter,messageListener);
        pullTaskExecutor.submit(this.pullBatchTask);
    }

    public void stop() {
        while(this.pullBatchTask.isRunning()) {
            try {
                Thread.sleep(1 * 1000);
            } catch (Throwable e) {
                //ignore
            }
        }
        this.pullBatchTask.stop();
        pullTaskExecutor.shutdown();
        executorService.shutdown();
        try {
            //等待重试任务结束
            while(executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                this.rertyMQProducer.shutdown();
                break;
            }
        } catch (Throwable e) {
            //igonre
        }
    }

    /**
     * 任务监听
     */
    static interface MessageListener {
        boolean consumer(List<MessageExt> msgs);
    }

    /**
     * 定时调度任务，例如每 10 分钟会被调度一次
     */
    class PullBatchTask implements Runnable {
        DefaultLitePullConsumer consumer;
        String consumerGroup;
        String nameserverAddr;
        String topic;
        String filter;
        private volatile boolean running = true;
        private MessageListener messageListener;

        public PullBatchTask(String consumerGroup, String nameserverAddr,String topic, String filter, 
                             MessageListener messageListener) {
            this.consumerGroup = consumerGroup;
            this.nameserverAddr = nameserverAddr;
            this.topic = topic;
            this.filter = filter;
            this.messageListener = messageListener;
            init();
        }

        private void init() {
            System.out.println("init 方法被调用");
            consumer = new DefaultLitePullConsumer(this.consumerGroup);
            consumer.setNamesrvAddr(this.nameserverAddr);
            consumer.setAutoCommit(true);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            try {
                consumer.subscribe(topic, filter);
                consumer.start();
            } catch (Throwable e) {
                e.printStackTrace();
            }

        }

        public void stop() {
            this.running = false;
            this.consumer.shutdown();
        }

        public boolean isRunning() {
            return this.running;
        }

        @Override
        public void run() {
            this.running = true;
            long startTime = System.currentTimeMillis() - 5 * 1000;
            System.out.println("run 方法被调用");
            int notFoundMsgCount = 0;

            while(running) {
                try {
                    // 拉取一批消息
                    List<MessageExt> messageExts = consumer.poll();
                    if(messageExts != null && !messageExts.isEmpty()) {
                        notFoundMsgCount = 0;//查询到数据，重置为 0；
                        // 使用一个业务线程池专门消费消息
                        try {
                            executorService.submit(new ExecuteTask(messageExts, messageListener));
                        } catch (RejectedExecutionException e) { //如果被拒绝，停止拉取，业务代码不去拉取，在
                            // RocketMQ 内部会最终也会触发限流，不会再拉取更多的消息，确保不会触发内存溢出。
                            boolean retry = true;
                            while (retry)
                            try {
                                Thread.sleep(5 * 1000);//简单的限流
                                executorService.submit(new ExecuteTask(messageExts, messageListener));
                                retry = false;
                            } catch (RejectedExecutionException e2) {
                                retry = true;
                            }
                        }

                        MessageExt last = messageExts.get(messageExts.size() - 1);
                        /**
                         * 如果消息处理的时间超过了该任务的启动时间，本次批处理就先结束
                         * 停掉该消费者之前，建议先暂停拉取，这样就不会从 broker 中拉取消息
                         * */
                        if(last.getStoreTimestamp() > startTime) {
                            System.out.println("consumer.pause 方法将被调用。");
                            consumer.pause(buildMessageQueues(last));
                        }

                    } else {
                        notFoundMsgCount ++;
                    }

                    //如果连续出现 5 次未拉取到消息，说明本地缓存的消息全部处理，并且 pull 线程已经停止拉取了,此时可以结束本次消
                    //息拉取，等待下一次调度任务
                    if(notFoundMsgCount > 5) {
                        System.out.println("已连续超过 5 次未拉取到消息，将退出本次调度");
                        break;
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
            this.running = false;
        }

        /**
         * 构建 MessageQueue
         * @param msg
         * @return
         */
        private Set<MessageQueue> buildMessageQueues(MessageExt msg) {
            Set<MessageQueue> queues = new HashSet<>();
            MessageQueue queue = new MessageQueue(msg.getTopic(), msg.getBrokerName(), msg.getQueueId());
            queues.add(queue);
            return queues;
        }
    }

    /**
     * 任务执行
     */
    class ExecuteTask implements Runnable {
        private List<MessageExt> msgs;
        private MessageListener messageListener;
        public ExecuteTask(List<MessageExt> allMsgs, MessageListener messageListener) {
            this.msgs = allMsgs.stream().filter((MessageExt msg) -> msg.getReconsumeTimes() <= 
                                                16).collect(Collectors.toList());
            this.messageListener = messageListener;
        }
        @Override
        public void run() {
            try {
                 this.messageListener.consumer(this.msgs);
            } catch (Throwable e) {
                //消息消费失败，需要触发重试
                //这里可以参考 PUSH 模式，将消息再次发送到服务端。
                try {
                    for(MessageExt msg : this.msgs) {
                        msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                        rertyMQProducer.send(msg);
                    }
                } catch (Throwable e2) {
                    e2.printStackTrace();
                    // todo 重试
                }
            }
        }
    }
}

// DefaultThreadFactory.java
package org.apache.rocketmq.example.simple.litepull;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class DefaultThreadFactory implements ThreadFactory {
    private AtomicInteger num = new AtomicInteger(0);
    private String prefix;

    public DefaultThreadFactory(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName(prefix + num.incrementAndGet());
        return t;
    }
}

// LitePullMain.java
package org.apache.rocketmq.example.simple.litepull;

import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.concurrent.*;

public class LitePullMain {
    public static void main(String[] args) {

        String consumerGroup = "dw_test_consumer_group";
        String nameserverAddr = "192.168.3.166:9876";
        String topic = "dw_test";
        String filter = "*";
        /** 创建调度任务线程池 */
        ScheduledExecutorService schedule = new ScheduledThreadPoolExecutor(1, new 
                                                 DefaultThreadFactory("main-schdule-"));
        schedule.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                BigDataPullConsumer demoMain = new BigDataPullConsumer(consumerGroup, nameserverAddr, topic, 
                                                                       filter);
                demoMain.registerMessageListener(new BigDataPullConsumer.MessageListener() {
                    /**
                     * 业务处理
                     * @param msgs
                     * @return
                     */
                    @Override
                    public boolean consumer(List<MessageExt> msgs) {
                        System.out.println("本次处理的消息条数：" + msgs.size());
                        return true;
                    }
                });
                demoMain.start();
                demoMain.stop();
            }
        }, 1000, 30 * 1000, TimeUnit.MILLISECONDS);

        try {
            CountDownLatch cdh = new CountDownLatch(1);
            cdh.await(10 , TimeUnit.MINUTES);
            schedule.shutdown();
        } catch (Throwable e) {
            //ignore
        }

    }
}

程序运行结果如下图所示：

符合预期，可以看到两次调度，并且每一次调度都正常结束。

首先对各个类的职责做一个简单介绍：

• MessageListener：用来定义用户的消息处理逻辑。
• PullBatchTask：使用 RocketMQ Lite Pull 消费者进行消息拉取的核心实现。
• ExecuteTask：业务处理任务，在内部实现调用业务监听器，并执行重试相关的逻辑。
• BigDataPullConsumer：本次业务的具体实现类
• LitePullMain：本次测试主入口类。

接下来对 PullBatchTask、ExecuteTask 的实现思路进行一个简单介绍，从而窥探一下消息 PULL 模式的一些使用要点。

PullBatchTask 的 run 方法主要是使用一个 while 循环，但通常不会用向 PUSH 模式实时监听，而是进行批量处理，即通过定时调度按批次进行处理，故需要有结束本次调度的逻辑，主要是为了提高消息拉取的效率，故本示例采用了本次任务启动只消费本次启动之前发送的消息，后面的新消息等聚集后在另一次调度时再消费，这里为了保证消费者停止时消息消费进度已经被持久化，这里并不会立即结束,而是在没有拉取合适的消息后调用 pause 方法暂停队列的消息，然后再连续多少次后并未拉取到消息后，在调用 DefaultLitePullConsumer 的 shutdown 方法，确保消息进度完整无误的提交到 Broker，从而避免大量消息重复消费。

消息消费端的业务处理这里引入了一个业务线程池，并且如果业务线程池积压，会触发消息拉取端的限流，从而避免内存溢出。

消息消费端在业务处理失败后，需要重试，将消息先发送到 Broker（主要的目的时方便消息消费进度向前推进）。