RocketMQ集群配置

 
RocketMQ 网络部署特点

• Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

• Broker部署相对复杂，Broker 分为Master与Slave，一个Master 可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定 义，BrokerId为0 表示Master，非0 表示Slave。Master也可以部署多个。每个 Broker与 Name Server集群中的所有节点建立长连接，定时注册 Topic信息到所有 Name Server。

• Producer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server取Topic 路由信息，并向提供Topic服务的Master 建立长连接，且定时向Master发送心跳。Producer完全无 状态，可集群部署。

• Consumer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server 取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker 配置决定。

分别启动

nohup sh mqnamesrv &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/namesrv.log

• 1
• 2
• 3

众所周知，RocketMQ有多种集群部署方式，它们的配置文件也是分开的，如下：

• 2m-noslave： 多Master模式
• 2m-2s-sync： 多Master多Slave模式，同步双写
• 2m-2s-async：多Master多Slave模式，异步复制

RocketMQ默认提供的配置文件都是最基本的，很多配置都是默认值，如下：

这些肯定不满足我们的要求，我们可以自己手动来配置

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a|broker-b
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/alibaba-rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/alibaba-rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/alibaba-rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/alibaba-rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/alibaba-rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/alibaba-rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536

#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=ASYNC_MASTER

#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62

Broker集群部署方式主要有以下几种：（Slave不可写，但可读）

这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用，不建议线上环境使用。

一个集群无 Slave，全是 Master，例如 2 个 Master 或者 3 个 Master。

优点：配置简单，单个Master 宕机或重启维护对应用无影响，在磁盘配置为 RAID10 时，即使机器宕机不可恢复情况下，由于RAID10 磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢）。性能最高。

缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到受到影响。

• 在192.168.1.101，启动第一个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.103，启动第二个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

每个 Master 配置一个 Slave，有多对Master-Slave，HA 采用异步复制方式，主备有短暂消息延迟，毫秒级。

优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，因为 Master 宕机后，消费者仍然可以从 Slave 消费，此过程对应用透明。不需要人工干预。性能同多 Master 模式几乎一样。

缺点：Master宕机，磁盘损坏情况，会丢失少量消息。

• 在192.168.1.101，启动第一个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.102，启动第一个 Slave（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.103，启动第二个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在机器 192.168.1.104，启动第二个 Slave（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

• 1

每个 Master 配置一个 Slave，有多对Master-Slave，HA 采用同步双写方式，主备都写成功，向应用才返回成功。

优点：数据与服务都无单点，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高

缺点：性能比异步复制模式略低，大约低10%左右，发送单个消息的 RT 会略高。目前主宕机后，备机不能自动切换为主机，后续会支持自动切换功能。

• 在192.168.1.101，启动第一个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.102，启动第一个 Slave（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.103，启动第二个 Master（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

• 在192.168.1.104，启动第二个 Slave（-n 192.168.1.101:9876 可省略）

nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties >$ROCKETMQ_HOME/log/mq.log >/dev/null 2>&1 &

tail -f -n 500 $ROCKETMQ_HOME/logs/rocketmqlogs/broker.log

• 1
• 2
• 3

以上 Broker 与 Slave 配对是通过指定相同的brokerName 参数来配对，Master 的 BrokerId必须是 0， 
Slave的BrokerId必须是大于 0 的数。另外一个 Master 下面可以挂载多个 Slave，同一 Master下的多个 Slave 通过指定不同的 BrokerId 来区分。

package com.somnus.rocketmq;

import java.util.concurrent.TimeUnit;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;

public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        /**
         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ProducerGroupName需要由应用来保证唯一,一类Producer集合的名称，这类Producer通常发送一类消息，
         * 且发送逻辑一致<br>
         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        final TransactionMQProducer producer = new TransactionMQProducer("ProducerGroupName");
        // nameserver服务
        producer.setNamesrvAddr("172.16.235.77:9876;172.16.235.78:9876");
        producer.setInstanceName("Producer");

        /**
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        producer.start();
        // 服务器回调Producer，检查本地事务分支成功还是失败
        producer.setTransactionCheckListener(new TransactionCheckListener() {

            public LocalTransactionState checkLocalTransactionState(
                    MessageExt msg) {
                System.out.println("checkLocalTransactionState --" + new String(msg.getBody()));
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        /**
         * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
         * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br>
         * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br>
         * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
         */

        for (int i = 0; i < 10; i++) {
            try {
                {
                    Message msg = new Message("TopicTest1", // topic
                            "TagA",                         // tag
                            "OrderID001",                   // key消息关键词，多个Key用KEY_SEPARATOR隔开（查询消息使用）
                            ("Hello MetaQA").getBytes());   // body
                    SendResult sendResult = producer.sendMessageInTransaction(
                            msg, new LocalTransactionExecuter(){
                                public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
                                    System.out.println("executeLocalTransactionBranch--msg=" + new String(msg.getBody()));
                                    System.out.println("executeLocalTransactionBranch--arg=" + arg);
                                    return LocalTransactionState.COMMIT_MESSAGE;
                                }
                            },
                            "$$$");
                    System.out.println(sendResult);
                }

                {
                    Message msg = new Message("TopicTest2", // topic
                            "TagB",                         // tag
                            "OrderID0034",                  // key 消息关键词，多个Key用KEY_SEPARATOR隔开（查询消息使用）
                            ("Hello MetaQB").getBytes());   // body
                    SendResult sendResult = producer.sendMessageInTransaction(
                            msg, new LocalTransactionExecuter(){
                                public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
                                    System.out.println("executeLocalTransactionBranch--msg=" + new String(msg.getBody()));
                                    System.out.println("executeLocalTransactionBranch--arg=" + arg);
                                    return LocalTransactionState.COMMIT_MESSAGE;
                                }
                            },
                            "$$$");
                    System.out.println(sendResult);
                }

                {
                    Message msg = new Message("TopicTest3", // topic
                            "TagC",                         // tag
                            "OrderID061",                   // key
                            ("Hello MetaQC").getBytes());   // body
                    SendResult sendResult = producer.sendMessageInTransaction(
                            msg, new LocalTransactionExecuter(){
                                public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
                                    System.out.println("executeLocalTransactionBranch--msg=" + new String(msg.getBody()));
                                    System.out.println("executeLocalTransactionBranch--arg=" + arg);
                                    return LocalTransactionState.COMMIT_MESSAGE;
                                }
                            },
                            "$$$");
                    System.out.println(sendResult);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            TimeUnit.MILLISECONDS.sleep(1000);
        }

        /**
         * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
         * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
         */
        // producer.shutdown();
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
            }
        }));
        System.exit(0);
    } // 执行本地事务，由客户端回调

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108
• 109
• 110
• 111
• 112
• 113
• 114
• 115
• 116
• 117
• 118
• 119
• 120
• 121
• 122

package com.somnus.rocketmq;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;  
import com.alibaba.rocketmq.common.message.MessageQueue;

public class Consumer {

    // Java缓存  
    private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();  

    /** 
     * 主动拉取方式消费 
     *  
     * @throws MQClientException 
     */  
    public static void main(String[] args) throws MQClientException {  
        /** 
         * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br> 
         * 注意：ConsumerGroupName需要由应用来保证唯一 ,最好使用服务的包名区分同一服务,一类Consumer集合的名称， 
         * 这类Consumer通常消费一类消息，且消费逻辑一致 
         * PullConsumer：Consumer的一种，应用通常主动调用Consumer的拉取消息方法从Broker拉消息，主动权由应用控制 
         */  
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("ConsumerGroupName");  
        // //nameserver服务  
        consumer.setNamesrvAddr("172.16.235.77:9876;172.16.235.78:9876");  
        consumer.setInstanceName("Consumber");  
        consumer.start();

        // 拉取订阅主题的队列，默认队列大小是4  
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");  
        for (MessageQueue mq : mqs) {  
            System.out.println("Consume from the queue: " + mq);  
            SINGLE_MQ: while (true) {  
                try {  
                    PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);  
                    List<MessageExt> list = pullResult.getMsgFoundList();  
                    if (list != null && list.size() < 100) {  
                        for (MessageExt msg : list) {  
                            System.out.println(new String(msg.getBody()));  
                        }  
                    }  
                    System.out.println(pullResult.getNextBeginOffset());  
                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());  
                    switch (pullResult.getPullStatus()) {
                        case FOUND:  
                            break;  
                        case NO_MATCHED_MSG:  
                            break;  
                        case NO_NEW_MSG:  
                            break SINGLE_MQ;  
                        case OFFSET_ILLEGAL:  
                            break;  
                        default:  
                            break;  
                    }  
                } catch (Exception e) {  
                    e.printStackTrace();  
                }  
            }  
        }  
        consumer.shutdown();  
    }  

    private static void putMessageQueueOffset(MessageQueue mq, long offset) {  
        offseTable.put(mq, offset);  
    }  

    private static long getMessageQueueOffset(MessageQueue mq) {  
        Long offset = offseTable.get(mq);  
        if (offset != null) {  
            System.out.println(offset);  
            return offset;  
        }  
        return 0;  
    }  
}