MQ之RocketMQ环境详细配置

wget https://mirror.bit.edu.cn/apache/rocketmq/4.7.1/rocketmq-all-4.7.1-source-release.zip

set "JAVA_OPT=%JAVA_OPT% -cp %CLASSPATH%" 

public class RocketMQDemo {
    static final String MQ_NAMESRVADDR = "127.0.0.1:9876";
    public static void main(String[] args) {
        // 分组名
        String groupName = "myGroup-1";
        // 主题名
        String topicName = "myTopic-1";
        // 标签名
        String tagName = "myTag-1";
        new Thread(() -> {
            try {
                producer(groupName, topicName, tagName);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (MQBrokerException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try {
                consumer(groupName, topicName, tagName);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }).start();
    }

    /**
     * @Description 生产者
     * @Author wanglei
     * @Param [groupName 分组名, topicName 主题名, tagName 标签名]
     **/
    public static void producer(String groupName, String topicName, String tagName) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(MQ_NAMESRVADDR);
        producer.start();
        String body = "Hello, World!";
        Message message = new Message(topicName, tagName, body.getBytes());
        producer.send(message);
        producer.shutdown();
    }

    /**
     * @Description 消费者
     * @Author wanglei
     * @Param [groupName 分组名, topicName 主题名, tagName 标签名]
     **/
    public static void consumer(String groupName, String topicName, String tagName) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(MQ_NAMESRVADDR);
        consumer.subscribe(topicName, tagName);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println(new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }
}

mqadmin.cmd updateTopic -b 192.168.1.108:10911 -n localhost:9876 -t myT

wget https://mirror.bit.edu.cn/apache/rocketmq/4.7.1/rocketmq-all-4.7.1-source-release.zip

yum list unzip：查看 unzip 包的安装情况
yum install unzip.x86_64：没有安装时，使用命令安装 unzip
rm rocketmq-all-4.7.1-source-release.zip   #解压后，可以删除原来的压缩包
mv rocketmq-all-4.7.1-source-release/ rocketmq-4.7.1    #对解压后的目录重命名

# 启动命令，并且常驻内存，nohup 属于后台启动，当前目录下生成 nohup.out 日志文件，也可以指定日志输出位置。

# 启动namesrv
# sh bin/mqnamesrv ：属于终端启动，直接输出日志信息，按 ctrl+c 可直接关闭退出
$ nohup sh bin/mqnamesrv &
# 查看启动日志，能看到 "The Name Server boot success" 字样则成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
#看到 “ The Name Server boot success. serializeType=JSON ” 表示启动成功

#启动broker
nohup sh bin/mqbroker -n localhost:9876 &

sh bin/mqbroker -n localhost:9876 ：属于终端启动，直接输出日志信息，按 ctrl+c 可直接关闭退出

tail -f ~/logs/rocketmqlogs/broker.log
#看到 "The broker[%s, 172.30.30.233:10911] boot success..." ,则表示启动成功

-- 关闭broker
sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
-- 关闭namesrv
sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

--查看集群情况    
./bin//mqadmin clusterList -n 127.0.0.1:9876
--查看 broker 状态    
./bin/mqadmin brokerStatus -n 127.0.0.1:9876 -b 172.20.1.138:10911
--查看 topic 列表    
./bin/mqadmin topicList -n 127.0.0.1:9876
--查看 topic 状态    
./bin/mqadmin topicStatus -n 127.0.0.1:9876 -t MyTopic (换成想查询的 topic)
--查看 topic 路由    
./bin/mqadmin topicRoute -n 127.0.0.1:9876 -t MyTopic