RocketMQ常用基本操作

admin2024-07-01  20

文章中的rabbitmq使用的是rocketmq-all-5.1.3-bin-release版本,需要安装包的可自行下载

RockerMQ启动停止命令

启动命令

nohup sh bin/mqnamesrv &

nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &

查看日志

tail -f ~/logs/rocketmqlogs/namesrv.log

tail -f ~/logs/rocketmqlogs/proxy.log

停止命令

sh bin/mqshutdown broker

sh bin/mqshutdown namesrv

集群状态

sh mqadmin clusterList -n 127.0.0.1:9876

RocketMQ常用基本操作,第1张

创建topic

sh mqadmin updateTopic -n 127.0.0.1:9876 rocket_test

查看所有topic信息

sh mqadmin topicList -n 127.0.0.1:9876

RocketMQ常用基本操作,第2张

sh mqadmin topicList -n 127.0.0.1:9876 -c

查看 Topic 路由信息

sh mqadmin topicRoute -n 127.0.0.1:9876 -t TopicTest

RocketMQ常用基本操作,第3张

发送测试消息

export NAMESRV_ADDR=localhost:9876

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

RocketMQ常用基本操作,第4张

消费消息

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

RocketMQ常用基本操作,第5张

Java代码收发消息

Producer

package com.rocket.demo;

import com.alibaba.fastjson.JSON;

import org.apache.rocketmq.client.producer.DefaultMQProducer;

import org.apache.rocketmq.client.producer.SendResult;

import org.apache.rocketmq.common.message.Message;

import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.HashMap;

import java.util.Map;

public class RocketProducerDemo {

    private final static String nameServer = "127.0.0.1:9876";

    private final static String producerGroup = "my_group2";

    // debezium-mysql-source-topic topic-test

    private final static String topic = "TopicTest";

    public static void main(String[] args) {

        try {

            // 初始化一个producer并设置Producer group name

            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);

//            DefaultMQProducer producer = new DefaultMQProducer();

            // 设置NameServer地址

            producer.setNamesrvAddr(nameServer);

            // 启动producer

            producer.start();

            for (int i = 0; i < 100; i++) {

                Map<String, String> data = new HashMap();

                data.put("id", i+"");

                data.put("name", i+","+System.currentTimeMillis());

                // 创建一条消息,并指定topic、tag、body等信息,tag可以理解成标签,对消息进行再归类,RocketMQ可以在消费端对tag进行过滤

                Message msg = new Message("TopicTest", "tagA", JSON.toJSONString(data).getBytes(RemotingHelper.DEFAULT_CHARSET));

                // 利用producer进行发送,并同步等待发送结果

                SendResult sendResult = producer.send(msg, 10000);

                System.out.println(sendResult);

            }

            // 一旦producer不再使用,关闭producer

            producer.shutdown();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

Consumer

package com.rocket.demo;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;

import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketConsumerDemo {

    public static void main(String[] args) throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_group");

        consumer.setNamesrvAddr("localhost:9876");

        // debezium-mysql-source-topic  topic-test debezium-mysql-source db-history-debezium-topic debezium-mysql-source

        consumer.subscribe("TopicTest", "*"); // 订阅主题和标签,* 表示订阅所有标签

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override

            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {

                for (MessageExt message : messages) {

                    System.out.println("Received message: " + new String(message.getBody()));

                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }

        });

        consumer.start();

        System.out.println("Consumer started");

    }

}

RocketMQ常用基本操作,第6张

常见问题

service not available now. It may be caused by one of the following reasons: the broker's disk is full [CL: 0.96 CQ: 0.96 INDEX: 0.96], messages are put to the slave, message store has been shut down

错误原因:博主测试的服务器磁盘使用率到0.96了,rocketmq不允许磁盘超过0.9,清理下磁盘数据即可

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明原文出处。如若内容造成侵权/违法违规/事实不符,请联系SD编程学习网:675289112@qq.com进行投诉反馈,一经查实,立即删除!