RocketMQ-核心篇(1)-软件安装

2024-02-21 01:34:32

RocketMQ安装指南：轻松掌握分布式消息队列系统

RocketMQ是一款备受推崇的开源分布式消息队列系统，以其高性能、高可用、高扩展性和高可靠性而闻名。如果您计划将RocketMQ集成到您的项目中，那么首先需要了解它的安装步骤。本指南将为您提供详细的安装说明，让您轻松上手RocketMQ。

步骤1：下载RocketMQ

访问RocketMQ官方网站，下载最新版本的安装包。您还可以从镜像网站获取安装包。下载完成后，将文件解压到您选择的目录中，例如/opt/rocketmq。

步骤2：安装RocketMQ

转到RocketMQ安装目录并执行以下命令：

sh bin/mqserver -n

您将看到一个配置界面，其中包含NameServer和Broker的设置。NameServer是RocketMQ的名称服务器，负责管理集群中的Broker节点，而Broker是RocketMQ的消息队列服务器。

根据您的需要配置设置，然后按回车键开始安装。完成后，您会看到如下输出：

Apache RocketMQ(3.6.2-SNAPSHOT) Startup Info:

步骤3：启动RocketMQ

要启动RocketMQ，请执行以下命令：

sh bin/mqnamesrv
sh bin/mqbroker -n

成功启动后，您会看到以下输出：

Starting broker OK

步骤4：停止RocketMQ

要停止RocketMQ，请执行以下命令：

sh bin/mqshutdown namesrv
sh bin/mqshutdown broker

停止后，您会看到以下输出：

Broker closed.

步骤5：常用RocketMQ命令

一旦安装并启动了RocketMQ，您可以使用以下命令对其进行管理：

bin/mqnamesrv：启动NameServer
bin/mqbroker -n：启动Broker
bin/mqshutdown namesrv：停止NameServer
bin/mqshutdown broker：停止Broker
bin/mqadmin -n：管理RocketMQ集群

代码示例

以下是使用Producer向RocketMQ主题发送消息的示例代码：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class ProducerExample {

    public static void main(String[] args) throws Exception {
        // 创建Producer
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer
        producer.start();

        // 创建消息
        Message message = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
        // 发送消息
        producer.send(message);
        // 关闭Producer
        producer.shutdown();
    }
}

以下是使用Consumer从RocketMQ主题接收消息的示例代码：

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.common.message.MessageExt;

import java.util.List;

public class ConsumerExample {

    public static void main(String[] args) throws Exception {
        // 创建Consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
        // 设置NameServer地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 订阅主题
        consumer.subscribe("TopicTest", "*");
        // 注册消息监听器
        consumer.registerMessageListener((List<MessageExt> messages, ConsumeConcurrentlyContext context) -> {
            for (MessageExt message : messages) {
                System.out.println(new String(message.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动Consumer
        consumer.start();
    }
}