Mac搭建单机版RocketMQ-4.9.3

文章目录

一、背景
二、RocketMQ
- 1、准备工作
- 2、方式一：项目源码启动
- - （1）启动nameServer
  - （2）启动broker
  - （3）Producer发布消息
  - （4）Consumer消费消息
  - （5）Dashboard查看
- 3、方式二：命令启动

一、背景

版本说明：

JDK：1.8
RocketMQ： 4.9.3
Dashboard：

二、RocketMQ 1、准备工作

下载地址：

https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.9.3/rocketmq-all-4.9.3-source-release.zip

解压RocketMQ的zip文件，并导入IDE。

本机环境准备：

（1）确认JAVA_HOME

（2）项目JDK选择1.8
若本地有多个JDK版本，记得选择1.8。

下面分两种方式启动：项目源码启动和命令启动。

2、方式一：项目源码启动

为了方便查看，新建rocketmq文件夹，之后创建conf、logs、store，如下图所示：

conf：将源码中distribution模块中的配置文件拷贝到这里。
logs：日志路径
store：存储相关内容。

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
# 添加此项，broker连接到本地的namesrv上
namesrvAddr = 127.0.0.1:9876
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
 
###########################################添加以下配置，路径请修改为自己的正确路径
# 配置存储位置，配置成上面新创建的路径。
storePathRootDir = /rocketmq/store
# commitlog 存储路径
storePathCommitLog = /rocketmq/store/commitlog
# 消费队列存储路径
storePathConsumeQueue = /rocketmq/store/commitlog/store/consumequeue
# 消息索引存储路径
storePathIndex = /rocketmq/store/index
# checkpoint文件存储路径
storeCheckPoint = /rocketmq/store/checkpoint
# abort文件存储路径
abortFile = /rocketmq/store/abort

在剩下两个logback文件中添加user.home配置，即新建的rocketmq路径。

（1）启动nameServer

启动类路径：org.apache.rocketmq.namesrv.NamesrvStartup。
编辑RunConfig，设置环境变量ROCKETMQ_HOME。

ROCKETMQ_HOME=/rocketmq

启动完成后，控制台输出：

（2）启动broker

启动类路径：org.apache.rocketmq.broker.BrokerStartup。
除了配置ROCKETMQ_HOME，还需要指定下使用server.conf。

-c /rocketmq/conf/broker.conf

启动成功：

（3）Producer发布消息

示例代码在example模块中，路径：org.apache.rocketmq.example.quickstart.Producer。

注意：
需要指定nameServer的地址。

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

        
        DefaultMQProducer producer = new DefaultMQProducer("ysx-group");
        //指定nameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
       
        producer.start();

        for (int i = 0; i < 5; i++) {
            try {

                Message msg = new Message("Topic-ysx" ,
                    "tag-test" ,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) 
                );

                
                SendResult sendResult = producer.send(msg);
                

                

                

                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        
        //producer.shutdown();
    }
}

控制台输出：

SendResult [sendStatus=SEND_OK, msgId=7F000001520A7F31245A4BDC23990000, offsetMsgId=C0A81FEF00002A9F000000000002F526, messageQueue=MessageQueue [topic=Topic-ysx, brokerName=broker-a, queueId=1], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=7F000001520A7F31245A4BDC23A60001, offsetMsgId=C0A81FEF00002A9F000000000002F5E8, messageQueue=MessageQueue [topic=Topic-ysx, brokerName=broker-a, queueId=2], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=7F000001520A7F31245A4BDC23A80002, offsetMsgId=C0A81FEF00002A9F000000000002F6AA, messageQueue=MessageQueue [topic=Topic-ysx, brokerName=broker-a, queueId=3], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=7F000001520A7F31245A4BDC23AA0003, offsetMsgId=C0A81FEF00002A9F000000000002F76C, messageQueue=MessageQueue [topic=Topic-ysx, brokerName=broker-a, queueId=0], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=7F000001520A7F31245A4BDC23AC0004, offsetMsgId=C0A81FEF00002A9F000000000002F82E, messageQueue=MessageQueue [topic=Topic-ysx, brokerName=broker-a, queueId=1], queueOffset=1]

（4）Consumer消费消息

示例代码在example模块中，路径：org.apache.rocketmq.example.quickstart.Consumer。

注意：
需要指定nameServer的地址。

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ysx-group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicTest", "*");

        
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List 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");
    }
}

控制台输出示例：

Consumer Started.
ConsumeMessageThread_ysx-producer-group_1 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=3, storeSize=194, queueOffset=251, sysFlag=0, bornTimestamp=1652606960700, bornHost=/192.168.31.239:60111, storeTimestamp=1652606960703, storeHost=/192.168.31.239:10911, msgId=C0A81FEF00002A9F000000000002F2E0, commitLogOffset=193248, bodyCRC=1250039395, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=252, CONSUME_START_TIME=1652611116478, UNIQ_KEY=7F00000151F37F31245A4BD9BC3C0002, CLUSTER=DefaultCluster, TAGS=tag-test}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 50], transactionId='null'}]] 
ConsumeMessageThread_ysx-producer-group_4 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=2, storeSize=194, queueOffset=251, sysFlag=0, bornTimestamp=1652606960698, bornHost=/192.168.31.239:60111, storeTimestamp=1652606960699, storeHost=/192.168.31.239:10911, msgId=C0A81FEF00002A9F000000000002F21E, commitLogOffset=193054, bodyCRC=1401636825, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=252, CONSUME_START_TIME=1652611116479, UNIQ_KEY=7F00000151F37F31245A4BD9BC3A0001, CLUSTER=DefaultCluster, TAGS=tag-test}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49], transactionId='null'}]] 
ConsumeMessageThread_ysx-producer-group_3 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=0, storeSize=194, queueOffset=252, sysFlag=0, bornTimestamp=1652606960704, bornHost=/192.168.31.239:60111, storeTimestamp=1652606960705, storeHost=/192.168.31.239:10911, msgId=C0A81FEF00002A9F000000000002F3A2, commitLogOffset=193442, bodyCRC=1032136437, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=253, CONSUME_START_TIME=1652611116481, UNIQ_KEY=7F00000151F37F31245A4BD9BC400003, CLUSTER=DefaultCluster, TAGS=tag-test}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 51], transactionId='null'}]] 
ConsumeMessageThread_ysx-producer-group_2 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=1, storeSize=194, queueOffset=251, sysFlag=0, bornTimestamp=1652606960689, bornHost=/192.168.31.239:60111, storeTimestamp=1652606960695, storeHost=/192.168.31.239:10911, msgId=C0A81FEF00002A9F000000000002F15C, commitLogOffset=192860, bodyCRC=613185359, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=253, CONSUME_START_TIME=1652611116481, UNIQ_KEY=7F00000151F37F31245A4BD9BC300000, CLUSTER=DefaultCluster, TAGS=tag-test}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48], transactionId='null'}]] 
ConsumeMessageThread_ysx-producer-group_5 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=1, storeSize=194, queueOffset=252, sysFlag=0, bornTimestamp=1652606960706, bornHost=/192.168.31.239:60111, storeTimestamp=1652606960706, storeHost=/192.168.31.239:10911, msgId=C0A81FEF00002A9F000000000002F464, commitLogOffset=193636, bodyCRC=601994070, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=253, CONSUME_START_TIME=1652611116482, UNIQ_KEY=7F00000151F37F31245A4BD9BC420004, CLUSTER=DefaultCluster, TAGS=tag-test}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 52], transactionId='null'}]]

（5）Dashboard查看

git：https://github.com/apache/rocketmq-dashboard
管理页面如下：

Cluster：

Topic：

Consumer：

Producer：

查看Message

可能出现的问题：
java.lang.RuntimeException: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 1 DESC: the producer group[ysx-producer-group] not exist BROKER: 127.0.0.1:10911
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/

Producer发送消息后，不执行shutdown操作即可。

3、方式二：命令启动

安装引导：

https://rocketmq.apache.org/docs/quick-start/

unzip rocketmq-all-4.9.3-source-release.zip
cd rocketmq-all-4.9.3/
mvn -Prelease-all -DskipTests clean install -U
cd distribution/target/rocketmq-4.9.3/rocketmq-4.9.3

Start Name Server

nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log

Start Broker

nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log

Send & Receive Messages
发送/接收消息时，需要先指定变量 NAMESRV_ADDR。

export NAMESRV_ADDR=localhost:9876
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

Mac搭建单机版RocketMQ-4.9.3

Java相关栏目本月热门文章