【RocketMQ】基于 RocketMQ 5.1.0 版本的定时消息实践

骑猪看日落 2023-09-30 20:48 29阅读 0赞

#### 文章目录 ####

*  搭建集群
 *   *  启动nameserver
     *  启动Broker
     *  启动dashboard
 *  功能测试
 *   *  代码
     *  结果
 *  性能测试
 *  Tips

## 搭建集群 ##

首先，你需要搭建一个RocketMQ最新版本的集群，为了方便测试功能、修改配置等，搭建1个nameserver和1个master即可，具体搭建方法就不细说了，简单说一下步骤：

### 启动nameserver ###

*  下载二进制包：https://rocketmq.apache.org/download，上传、解压

![81eb874611084d2ea9d63f3eed736ab1.png][]

*  创建nameserver配置，启动nameserver：
    
    如果你需要修改日志保存路径，需要在conf目录下，找到对应的logback配置文件，修改所有路径相关的配置项，后面的Broker也一样。
    
    nameserver.conf：
    
        serverChannelMaxIdleTimeSeconds=120
        listenPort=19876
        serverSocketSndBufSize=65535
        serverAsyncSemaphoreValue=64
        serverCallbackExecutorThreads=0
        rocketmqHome=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0
        clusterTest=false
        serverSelectorThreads=5
        useEpollNativeSelector=false
        orderMessageEnable=false
        serverPooledByteBufAllocatorEnable=true
        serverWorkerThreads=8
        kvConfigPath=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0/store/kvConfig.json
        serverSocketRcvBufSize=65535
        productEnvName=center
        serverOnewaySemaphoreValue=256
        configStorePath=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0/conf/nameserver.conf
    
    启动脚本：
    
        nohup sh /neworiental/rocketmq-5.1.0/rocketmq-5.1.0/bin/mqnamesrv -c /neworiental/rocketmq-5.1.0/rocketmq-5.1.0/conf/nameserver.conf >/dev/null 2>&1 &

### 启动Broker ###

*  修改Broker配置，启动Broker：
    
    broker.conf
    
        brokerClusterName = 5-1-0-Cluster
        brokerName = broker-a
        brokerId = 0
        deleteWhen = 04
        fileReservedTime = 48
        brokerRole = ASYNC_MASTER
        flushDiskType = ASYNC_FLUSH
        autoCreateTopicEnable=true
        autoCreateSubscriptionGroup=true
        maxTransferBytesOnMessageInDisk=65536
        listenPort=3140
        namesrvAddr=172.24.30.192:19876;
        rocketmqHome=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0
        storePathConsumerQueue=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0/store/consumequeue
        brokerIP2=172.24.30.194
        brokerIP1=172.24.30.194
        aclEnable=false
        storePathRootDir=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0/store
        storePathCommitLog=/neworiental/rocketmq-5.1.0/rocketmq-5.1.0/store/commitlog
    
    启动脚本：
    
        #!/bin/bash
        . /etc/profile
        
        PID=`ps -ef | grep 'rocketmq-5.1.0' | grep -v grep | awk '{print $2}'`
        if [[ "" !=  "$PID" ]]; then
          echo "killing rocketmq-5.1.0 : $PID"
          kill $PID
        fi
        
        sleep 1
        
        nohup sh /neworiental/rocketmq-5.1.0/rocketmq-5.1.0/bin/mqbroker -c /neworiental/rocketmq-5.1.0/rocketmq-5.1.0/conf/broker.conf >/dev/null 2>&1 &
        echo "deploying rocketmq-5.1.0..."

### 启动dashboard ###

这一步，按需处理

下载地址：https://github.com/apache/rocketmq-dashboard

克隆之后，打成jar包，上传，编写启动脚本：

MAIN_JAR="-jar /neworiental/rocketmq-5.1.0/dashboard/rocketmq-console-ng-2.0.0.jar  "
    JAVA_ARGS="-server -Xms4g -Xmx4g -XX:NewSize=512m -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=70 -XX:+PrintGCDetails -XX:+PrintHeapAtGC  -XX:ThreadStackSize=512 -Xloggc:${LOGS_DIR}/gc.log "
    
    CONSOLE_ARGS="--server.port=18281 --rocketmq.config.loginRequired=false --rocketmq.config.namesrvAddr=172.24.30.192:19876 "
    
    if [ ! -d ${LOGS_DIR} ]
    then
      mkdir -p ${LOGS_DIR}
    fi
    
    echo ${JAVA_ARGS} ${MAIN_JAR} ${CONSOLE_ARGS}
    nohup java ${JAVA_ARGS} ${MAIN_JAR} ${CONSOLE_ARGS} >/dev/null 2>&1 &
    echo "deploying rocketmq-dashboard now ..."

搭建完成，打开dashboard

![3d248a08f5af2eb64258eb525af77cd3.png][]

## 功能测试 ##

### 代码 ###

测试定时/延时消息的功能，需要使用最新版本的客户端：

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>5.1.0</version>
            </dependency>

当然，官方将常见语言的客户端，单独整合了一个开源项目，我觉得不是很习惯，pass。项目地址：https://github.com/apache/rocketmq-clients 感兴趣的可以试试。

**生产者**：

对于生产者，不需要做太多改动，只需要为消息加一个延时或定时的参数即可：

public static void main(String[] args) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer("test-producer", false);
            producer.setNamesrvAddr("172.24.30.192:19876");
            producer.start();
            for (int i = 0; i < 10; i++) {
        
                Message msg = new Message();
                msg.setTopic("delay-topic");
                msg.setBody("这是一条延迟消息".getBytes(RemotingHelper.DEFAULT_CHARSET));
                Duration messageDelayTime = Duration.ofSeconds(10);
                long delayTimestamp = System.currentTimeMillis() + messageDelayTime.toMillis();
                // 绝对时间：定时消息
                msg.setDeliverTimeMs(delayTimestamp);
                // 相对时间：延时消息
    //            msg.setDelayTimeSec(1000 * 5);
                SendResult sendResult = producer.send(msg);
                System.out.printf("发送时间：%s %n", formatter.format(LocalDateTime.now()));
            }
            producer.shutdown();
        }

**消费者**：

消费者和消费普通消息一样，无需特殊设置：

public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-test-0317", false);
            //设置nameserver
            consumer.setNamesrvAddr("172.24.30.192:19876");
            //设置topic,subExpression即设置订阅的tag，*表示所有
            consumer.subscribe("delay-topic", "*");
            //从最新的offset拉取
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //注册监听
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
        
                System.out.printf("接收时间：%s %n", formatter.format(LocalDateTime.now()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }

### 结果 ###

生产者：

![5f5aeb9f73a39093b2934d93891863d9.png][]

消费者：

![1cbfd4ea47188ac3d38693428ee8058f.png][]

误差大概为几百毫秒。

## 性能测试 ##

官方提供了压测代码，org.apache.rocketmq.example.benchmark.timer.TimerProducer

需要下载最新版本的源码，下面是单Broker、100个线程、消息大小为1kb情况下的TPS情况：

![f7da7d211b9e1044751fc3506abc4b22.png][]

如果不想下载源码，可以使用下面的代码做简单测试，从dashboard观察TPS情况：

package cn.xdf.xadd.rmq.test.producer;
    
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.ThreadFactoryImpl;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.io.UnsupportedEncodingException;
    import java.time.Duration;
    import java.time.format.DateTimeFormatter;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class TestProducer {
        
    
        private static final String MSG = StringUtils.repeat('a', 1024);
        private static final String TOPIC = "delay-topic";
        private static final Integer THREAD_COUNT = 100;
        private static final ExecutorService SEND_THREAD_POOL = new ThreadPoolExecutor(
                THREAD_COUNT,
                THREAD_COUNT,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactoryImpl("ProducerSendMessageThread_"));
    
        static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS");
    
        static DefaultMQProducer producer = new DefaultMQProducer("test-producer", false);
    
    
        public static void main(String[] args) throws Exception {
        
            producer.setNamesrvAddr("172.24.30.192:19876");
            producer.start();
            benchmark();
        }
    
        private static void benchmark() {
        
            for (int i = 0; i < THREAD_COUNT; i++) {
        
                SEND_THREAD_POOL.execute(() -> {
        
                    for (int j = 1; j <= 10000; j++) {
        
                        sendMsg();
                    }
                });
            }
        }
    
        private static void sendMsg() {
        
            try {
        
                producer.send(buildMessage());
            } catch (Exception e) {
        
                System.out.println(e);
            }
        }
    
        private static Message buildMessage() throws UnsupportedEncodingException {
        
            Message msg = new Message();
            msg.setTopic(TOPIC);
            msg.setBody(MSG.getBytes(RemotingHelper.DEFAULT_CHARSET));
            Duration messageDelayTime = Duration.ofHours(10);
            long delayTimestamp = System.currentTimeMillis() + messageDelayTime.toMillis();
            msg.setDeliverTimeMs(delayTimestamp);
            return msg;
        }
    
    }

## Tips ##

1.  默认支持最大延迟时间为3天，可以根据broker配置：timerMaxDelaySec 修改；
2.  timerPrecisionMs 默认为1000，改小可提高精度，建议使用默认；
3.  对于超过时间轮槽位时间的消息，会通过取模的方式复用槽位，所以支持超过7天的延迟消息；
4.  消息积压不会导致OOM，时间轮数据在内存中用的是DirectByteBuffer，用的直接内存；
5.  配置项：timerEnableDisruptor 默认为关闭，按理说开启能提高性能，降低延迟（试了一下，改为true不生效，会被重置为false）

[81eb874611084d2ea9d63f3eed736ab1.png]: https://img-blog.csdnimg.cn/img_convert/81eb874611084d2ea9d63f3eed736ab1.png
[3d248a08f5af2eb64258eb525af77cd3.png]: https://img-blog.csdnimg.cn/img_convert/3d248a08f5af2eb64258eb525af77cd3.png
[5f5aeb9f73a39093b2934d93891863d9.png]: https://img-blog.csdnimg.cn/img_convert/5f5aeb9f73a39093b2934d93891863d9.png
[1cbfd4ea47188ac3d38693428ee8058f.png]: https://img-blog.csdnimg.cn/img_convert/1cbfd4ea47188ac3d38693428ee8058f.png
[f7da7d211b9e1044751fc3506abc4b22.png]: https://img-blog.csdnimg.cn/img_convert/f7da7d211b9e1044751fc3506abc4b22.png