RocketMQ集群配置

  1. 云栖社区>
  2. 博客>
  3. 正文

RocketMQ集群配置

追风筝的猪 2018-03-23 00:19:13 浏览934
展开阅读全文

这里写图片描述

这里写图片描述 
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 配置决定。

这里写图片描述

NameServer集群 IP地址
NameServer-1 192.168.1.101
NameServer-2 192.168.1.102

分别启动

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。

brokerName brokerId brokerRole IP地址
broker-a 0 ASYNC_MASTER 192.168.1.101
broker-b 0 ASYNC_MASTER 192.168.1.103

优点:配置简单,单个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 采用异步复制方式,主备有短暂消息延迟,毫秒级。

brokerName brokerId brokerRole IP地址
broker-a 0 ASYNC_MASTER 192.168.1.101
broker-a 1 SLAVE 192.168.1.102
broker-b 0 ASYNC_MASTER 192.168.1.103
broker-b 1 SLAVE 192.168.1.104

优点:即使磁盘损坏,消息丢失的非常少,且消息实时性不会受影响,因为 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 采用同步双写方式,主备都写成功,向应用才返回成功。

brokerName brokerId brokerRole IP地址
broker-a 0 SYNC_MASTER 192.168.1.101
broker-a 1 SLAVE 192.168.1.102
broker-b 0 SYNC_MASTER 192.168.1.103
broker-b 1 SLAVE 192.168.1.104

优点:数据与服务都无单点,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;  
    }  
}

网友评论

登录后评论
0/500
评论
追风筝的猪
+ 关注