首页 RocketMQ 系列

# 搭建 RocketMQ 集群

根据之前的了解，我们已经知道了 RocketMQ 可以单机部署，支持集群部署。

RocketMQ 支持以下集中部署方式 1.单Master模式部署 ， 2.多Master模式部署 ， 3.多Master多Slaver模式部署

# 单 Master 模式

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

# 启动 NameServer

1
2
3
4
5
6
### 首先启动Name Server
$ nohup sh mqnamesrv &
 
### 验证Name Server 是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

# 启动 Broker

1
2
3
4
5
6
### 启动Broker
$ nohup sh bin/mqbroker -n localhost:9876 &

### 验证Name Server 是否启动成功，例如Broker的IP为：192.168.1.2，且名称为broker-a
$ tail -f ~/logs/rocketmqlogs/broker.log 
The broker[broker-a, 192.169.1.2:10911] boot success...

# 验证

我们使用 Java 程序来简单的验证下消息的发送和消费，这也是我们第一次去使用 RocketMQ.

# 加入 maven 依赖

1
2
3
4
5
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.3.0</version>
</dependency>

# 生产消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class Producer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
    	producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
    	    // 创建消息，并指定Topic，Tag和消息体
    	    Message msg = new Message("TopicTest" /* Topic */,
        	"TagA" /* Tag */,
        	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

# 消费消息

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
public class Consumer {

	public static void main(String[] args) throws InterruptedException, MQClientException {

    	// 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");

    	// 设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");

    	// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
    	// 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
	}
}

遇到的问题:

按照上面的步骤，你运行成功了吗？？？

我反正是没有成功的生产消息。为什么呢？

1.No route info of this topic, TopicTest

原因是：我们在启动的时候，并没有创建 TopicTest 这个 Topic.

1
2
3
4
5
6
7
Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest
See http://rocketmq.apache.org/docs/faq/ for further details.
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:610)
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1223)
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1173)
	at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:214)
	at com.fxb.learn_rocketmq.Producer.main(Producer.java:23)

这种问题的原因两种。 1. 网络不通，连接不到端口。 2. 没有创建这个 Topic.
真不巧，我全遇到了。

第一种呢，使用 telnet 命令，如果通了就解决了。
第二种，使用 sh mqadmin updateTopic -n localhost:9876 -b 192.168.1.63:10911 -t TopicTest -n 表示 NameSerer 的地址，-b 表示 Broker 的地址，-t
后面的就是 topicName 了。

2.Broker 启动失败，提示 空间不足。

原因是：在 runServer.sh 和 runbroker.sh 脚本中配置了启动的内存大小， NameServer 启动的堆内存是 4G, 而 broker 启动的是 8g 所以，回报错因 空间不足，启动失败。

3. 运行 mqadmin 命令是报错

我遇到的问题是因为 宿主机安装的是 openJDK, 导致 脚本没有加载到对应的 ext 包，所以在 tools.sh 中加入 ext / 包的路径就可以了。 ext/ 就是 jre 下的 ext 包。

如下图.

# Rocket 运行程序

我们先不急着去学习其他的部署方式，我们现在去看下 Rocket 运行程序的目录是什么样的。目的是了解 RocketMQ 是怎么跑起来的。这样遇到问题的时候，我们可以顺藤摸瓜去排查问题是出在什么地方。

  • benchmark 包：提供了 Rocket 用于基准测试的工具。
    producer.sh ：测试普通 MQ 生产者的性能。
    consumer.sh ：测试 MQ 消费者的性能。
    tproducer.sh ：测试事务 MQ 生产者的性能。
    batchproducer.sh: 测试批量生产者的性能。
  • bin: 提供了 启动关闭 NameServer, 启动关闭 Broker，管理 MQ 的工具。
    • mqnamesrv： 启动 NameServer 的脚本
    • mqbroker: 启动 Broker 的脚本
    • mqadmin: 管理 RocketMQ 的脚本
    • mqshutdown: 关闭 MQ 的脚本
  • conf: 对 RocketMQ 进行配置的文件目录。
    • broker.conf： broker 的配置
    • logback-*.xml: logback 日志的配置，比如 Broker，nameServer,tools 的日志配置
    • plain_acl.yml： 权限管理的文件
    • tools.yml: 脚本工具的配置
    • dledger: 容灾相关的配置
  • lib: 这里存放的就是编译之后的 class 文件。

如果在启动或者使用其他的命令遇到问题的时候，可以到对应的目录中，找到文件，然后看下代码进行处理。

下面我们接着去看其他的部署方式。

# 多 Master 模式

一个集群无 Slave，全是 Master，例如 2 个 Master 或者 3 个 Master，这种模式的优缺点如下：

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

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

# 启动 NameServer

NameServer 需要先于 Broker 启动，且如果在生产环境使用，为了保证高可用，建议一般规模的集群启动 3 个 NameServer，各节点的启动命令相同，如下：

1
2
3
4
5
6
### 首先启动Name Server
$ nohup sh mqnamesrv &
 
### 验证Name Server 是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

# 启动 Broker 集群

1
2
3
4
5
6
7
### 在机器A，启动第一个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties &
 
### 在机器B，启动第二个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties &

...

如上启动命令是在单个 NameServer 情况下使用的。对于多个 NameServer 的集群，Broker 启动命令中 -n 后面的地址列表用分号隔开即可，例如 192.168.1.1:9876;192.161.2:9876 。

# 1.3 多 Master 多 Slave 模式 - 异步复制

每个 Master 配置一个 Slave，有多对 Master-Slave，HA 采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

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

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

# 启动 NameServer

1
2
3
4
5
6
### 首先启动Name Server
$ nohup sh mqnamesrv &
 
### 验证Name Server 是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

# 启动 Broker 集群

1
2
3
4
5
6
7
8
9
10
11
### 在机器A，启动第一个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties &
 
### 在机器B，启动第二个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties &
 
### 在机器C，启动第一个Slave，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties &
 
### 在机器D，启动第二个Slave，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties &

# 多 Master 多 Slave 模式 - 同步双写

每个 Master 配置一个 Slave，有多对 Master-Slave，HA 采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

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

  • 缺点：性能比异步复制模式略低（大约低 10% 左右），发送单个消息的 RT 会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

# 启动 NameServer

1
2
3
4
5
6
### 首先启动Name Server
$ nohup sh mqnamesrv &
 
### 验证Name Server 是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

# 启动 Broker 集群

1
2
3
4
5
6
7
8
9
10
11
### 在机器A，启动第一个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties &
 
### 在机器B，启动第二个Master，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties &
 
### 在机器C，启动第一个Slave，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties &
 
### 在机器D，启动第二个Slave，例如NameServer的IP为：192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties &

以上 Broker 与 Slave 配对是通过指定相同的 BrokerName 参数来配对，Master 的 BrokerId 必须是 0，Slave 的 BrokerId 必须是大于 0 的数。另外一个 Master 下面可以挂载多个 Slave，同一 Master 下的多个 Slave 通过指定不同的 BrokerId 来区分。$ROCKETMQ_HOME 指的 RocketMQ 安装目录，需要用户自己设置此环境变量。

# mqadmin 管理工具

关于 mqadmin 命令的使用，我们可以通过 sh mqadmin 命令来查看命令的选项，用的多自然就记住了，后面我们也不会通过这种方式来管理 RocketMQ 集群。但是在我们学习的过程中，我们还是会非常频繁的使用这个命令。

注意：

  1. 执行命令方法： ./mqadmin {command} {args}
  2. 几乎所有命令都需要配置 - n 表示 NameServer 地址，格式为 ip:port
  3. 几乎所有命令都可以通过 - h 获取帮助
  4. 如果既有 Broker 地址（-b）配置项又有 clusterName（-c）配置项，则优先以 Broker 地址执行命令，如果不配置 Broker 地址，则对集群中所有主机执行命令，只支持一个 Broker 地址。-b 格式为 ip:port，port 默认是 10911
  5. 在 tools 下可以看到很多命令，但并不是所有命令都能使用，只有在 MQAdminStartup 中初始化的命令才能使用，你也可以修改这个类，增加或自定义命令
  6. 由于版本更新问题，少部分命令可能未及时更新，遇到错误请直接阅读相关命令源码

# 最后

期望和你一起遇见更好的自己

更新于 阅读次数 次

请我喝[咖啡]~(￣▽￣)~*

方小白 微信支付

微信支付

方小白 支付宝

支付宝

方小白 numberpay

numberpay