最近在学习消息队列, 之前在window上搭建了一次, 这次在linux上搭建rocketmq出现了不少问题, 特此记录

先检查Linux上的rocketmq

查看后台是否存在进程的运行：

ps －ef | grep mq 

另外一个选择是直接查找

sudo find / －name mqbroker

如果存在，进行版本更新（删除）操作

直接删文件夹 rm －r 删除

修改环境变量

解压新rocketmq

unzip解压到新目录

unzip -d /usr/local/rocketmq rocketmq-all-4.9.0-bin-release.zip

创建日志目录

创建目录

在Rocketmq目录下执行

mkdir -p store store/commitlog store/consumequeue

注意需要再在broker.conf中编辑

listenPort=10911

namesrvAddr=localhost:9876

storePathRootDir=/usr/local/rocketmq/store

storePathCommitLog=/usr/local/rocketmq/store/commitlog

storePathConsumerQueue=/usr/local/rocketmq/store/consumequeue

修改runbroker和runnameserver大小

全部改成512或者256m [ 初始是8G或者4G ]

修改配置文件

在 sudo gedit /etc/profile 中进行修改

# rocketmqPath

ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.9.0-bin-release

PATH=$PATH:$ROCKETMQ_HOME/bin

export ROCKETMQ_HOME PATH

改完记得重启!

检查防火墙

一般是关闭的, 没有关闭就关闭掉

sudo ufw status

启动与测试

在rocketmq-all-4.9.0-bin-release/bin中, 执行代码:

nohup sh mqnamesrv &

nohup sh mqbroker -c /usr/local/rocketmq/rocketmq-all-4.9.0-bin-release/conf/broker.conf &

如果测试用例失败执行这个

nohup sh mqbroker -n localhost:9876 autoCreateTopicEnable=true &

nohup sh mqbroker -n localhost:9876 autoCreateTopicEnable=true &

检查运行

1. 使用jps命令

2. sh mqadmin clusterList

3. 查看日志

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

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

测试用例

在电脑上, 使用idea创建简单的消费者和生产者

在配置文件里面对应到Rocketmq的接口即可

检查ip

linux中是 ifconfig

windows中是 ipconfig

Rocketmq端口都是 9876

生产者实例

package org.example;
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;

public class SimpleProducer {
    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("192.168.33.130: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();
    }

}

消费者实例

package org.example;
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.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

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

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");

        // Specify name server addresses.
        consumer.setNamesrvAddr("192.168.33.130: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");
    }
}

直接运行即可查看

测试完成后关闭Rocketmq

sh mqshutdown namesrv
sh mqshutdown broker