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（主要的目的时方便消息消费进度向前推进）。