转载

在开发环境下，基于Springboot的RocketMQ示例（含安装步骤、错误分析）

在看这文章之前建议先看看先前架构原理介绍文章：

【必须先理解的RocketMQ入门手册，才能再次深入解读】

RocketMQ服务器启动

linux环境

下载编译源码

# 下载$ 
  > wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source- > 
  # 解压$
  >unzip rocketmq-all-4.7.0-source-release.zip
  > cd rocketmq-all-4.7.0/
  # 编译$
  > mvn -Prelease-all -DskipTests clean install -U
  > cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0

启动 Name Server

# 启动 Name Server 服务
 > nohup sh bin/mqnamesrv &
 # 启动完成后，查看日志$
 > tail -f ~/logs/rocketmqlogs/namesrv.log
  The Name Server boot success...

启动 Broker

在 conf 目录下，RocketMQ 提供了多种 Broker 的配置文件：

broker.conf
2m/
2m-2s-async/
2m-2s-sync/
dledger/

# 启动 Broker服务
 > nohup sh bin/mqbroker -n localhost:9876 &
 # 启动完成后，查看日志$
 > tail -f ~/logs/rocketmqlogs/broker.log 
  The broker[%s, 172.30.30.233:10911] boot success...

其中，参数：

-c
-n

关闭服务器

> sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK

> sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664

windows环境

首先去官网下载编译之后的版本,然后解压到本地目录

官网链接： http://rocketmq.apache.org/do...
下载目标：Binary: [rocketmq-all-4.7.0-bin-release.zip
配置ROCKETMQ_HOME到系统环境变量中,启动脚本将读取ROCKETMQ_HOME变量
分别进入bin目录下启动如下脚本(需要设置内存参数，防止内存过大，启动失败，具体看< 常出现的错误 >小节):

3.1 启动namesrv

运行命令：
```
mqnamesrv.cmd
```
log : The Name Server boot success. serializeType=JSON

3.2 启动brokerserver

运行命令：
```
mqbroker.cmd -n localhost:9876
```
log : The broker[IQSZ-L01898, 10.111.45.111:10911] boot success. serializeType=JSON and name server is localhost:9876

关闭服务器

log：killing name server
mqshutdown.cmd namesrv
log：killing broker

RocketMQ发送消息和消费消息

RocketMQ发送消息和消费消息，先启动消费者，然后再启动生产者

添加依赖

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

发送消息

发送消息--同步

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
            DefaultMQProducer("test-group");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

发送消息--异步

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("test—group");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        for (int i = 0; i < 100; i++) {
                final int index = i;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }

发送消息--单向

public class OnewayProducer {
    public static void main(String[] args) throws Exception{
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            producer.sendOneway(msg);

        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

消费消息

public class Consumer {

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

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-group");
         
        // Specify name server addresses.
        consumer.setNamesrvAddr("localhost:9876");
        
        // Subscribe one more more topics to consume.
        consumer.subscribe("TopicTest", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        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;
            }
        });

        //Launch the consumer instance.
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }

常出现的错误

安装中出现的错误

防止内存设置过大

修改 runbroker.cmd

配置文件

set "JAVA_OPT=%JAVA_OPT% -server -Xms500m -Xmx500m -Xmn500m"

set "JAVA_OPT=%JAVA_OPT% -XX:MaxDirectMemorySize=1g"

在开发环境下，基于Springboot的RocketMQ示例（含安装步骤、错误分析）

修改 runserver.cmd

配置文件

set "JAVA_OPT=%JAVA_OPT% -server -Xms500m -Xmx500m -Xmn500m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

在开发环境下，基于Springboot的RocketMQ示例（含安装步骤、错误分析）

启动NAMESERVER报错

unrecognized vm option 'MetasoaceSize=128m'

解决方法：更换jdk版本为1.8即可

启动BROKER报错

错误: 找不到或无法加载主类 xxxxxx’

解决方法:打开runbroker.cmd(windows)，然后将‘%CLASSPATH%’加上英文双引

使用过程中出现的错误

No route info of this topic

Broker禁止自动创建Topic，且用户没有通过手工方式创建Topic

查看是否允许自动创建topic

命令：mqbroker.cmd -n localhost:9876 -p

mq开启自动创建topic参数

命令：mqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true
Broker 没有正确连接到 Name Server

查看broker.log日志

位置： /安装目录/conf/logback_broker.xml中日志位置

日志信息：broker.log

日志信息：namesrv.log
Producer 没有正确连接到 Name Server

linux环境：查询防火墙是否通

错误分析方法

日志分析法：

查看broker日志
- 关注broker是否有注册到nameserver
  
  register broker to name server localhost:9876 OK
- 关注生产者是否连接到broker
```
new producer connected, group: test_group channel: ClientChannelInfo ...
```
- 查看已经创建的topic是否包含自己想要的topic
  
  2020-04-21 15:58:22 INFO main - load exist local topic, TopicConfig [topicName=test_group, readQueueNums=1, writeQueueNums=1, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false]
- 查看消费者是否连接到broker
  
  new consumer connected, group: test_group CONSUME_PASSIVELY CLUSTERING channel: ClientChannelInfo ...
查看nameserver日志
- 关注broker是否注册到nameserver
  
  new broker registered, localhost:10911
- 查看topic消息
  
  ...