在Linux中用源码安装RocketMQ-不过是从头再来罢了

1.下载

官网下载地址：https://archive.apache.org/dist/rocketmq/

比如我下载的 rocketmq-all-5.2.0-bin-release.zip.

2.上传与解压

# cd /usr/local/soft

# mkdir /usr/local/rocketmq

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

3.配置启动参数

1.修改jvm参数

在 /usr/local/rocketmq/rocketmq-all-5.2.0-bin-release/bin目录中

runserver.sh
runbroker.sh
tools.sh

将上面三个启动脚本的 JAVA_OPT项的jvm参数改为 ‐Xms512m ‐Xmx512m ‐Xmn512m

2.修改broker.conf
在broker.conf文件最下面添加如下配置：

# cd /usr/local/rocketmq/rocketmq-all-5.2.0-bin-release/conf

# vim broker.conf

### 修改内容
brokerIP = 116.14.0.50

若是云服务器,则brokerIP = 公网IP,若是Local则不用配置。

4.启动NameServer

安装完RocketMQ包后，我们启动NameServer

4.1 启动namesrv

# cd /usr/local/rocketmq/rocketmq-all-5.2.0-bin-release

# nohup sh bin/mqnamesrv &

验证namesrv是否启动成功

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

The Name Server boot success...

我们可以在namesrv.log 中看到 'The Name Server boot success..'，表示NameServer 已成功启动。

4.2 启动Broker+Proxy

NameServer成功启动后，我们启动Broker和Proxy。这里我们使用 Local 模式部署，即 Broker 和 Proxy 同进程部署。

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

验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a

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

The broker[broker-a,192.168.137.150:10911] boot success...

我们可以在 proxy.log 中看到“The broker[brokerName,ip:port] boot success..”，这表明 broker 已成功启动。

4.3 关闭服务

# 关闭服务
# sh mqshutdown namesrv
# sh mqshutdown broker

5.测试

5.1 工具测试消息收发

在进行工具测试消息收发之前，我们需要告诉客户端NameServer的地址，RocketMQ有多种方式在客户端中设置NameServer地址，这里我们利用环境变量NAMESRV_ADDR

# export NAMESRV_ADDR=localhost:9876
# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
SendResult [sendStatus=SEND_OK, msgId= ...

# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_%d Receive New Messages: [MessageExt...

5.2 SDK测试消息收发

工具测试完成后，我们可以尝试使用 SDK 收发消息。
这里以 Java SDK 为例介绍一下消息收发过程。

5.2.1 创建一个Java工程

5.2.2 添加Java依赖库

在 pom.xml 文件中添加以下依赖引入Java依赖库.

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>5.0.6</version>
</dependency>

5.2.3 创建 Topic

通过mqadmin创建 Topic。

$ sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster

创建Topic

主题必须有消息类型，且只能接收相对应消息类型的消息

Normal：普通消息，消息本身无特殊语义，消息之间也没有任何关联。
FIFO：顺序消息，Apache RocketMQ 通过消息分组MessageGroup标记一组特定消息的先后顺序，可以保证消息的投递顺序严格按照消息发送时的顺序。
Delay：定时/延时消息，通过指定延时时间控制消息生产后不要立即投递，而是在延时间隔后才对消费者可见。
Transaction：事务消息，Apache RocketMQ 支持分布式事务消息，支持应用数据库更新和消息调用的事务一致性保障。

普通消息

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=NORMAL

定时/延时消息

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=DELAY

顺序消息

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=FIFO

事务消息

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=TRANSACTION

nameserver_address是9876端口,
cluster_name一般为DefaultCluster

5.2.4 编写生产者

在已创建的Java工程中，创建发送普通消息程序并运行，示例代码如下：

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerExample {
    private static final Logger logger = LoggerFactory.getLogger(ProducerExample.class);

    public static void main(String[] args) throws ClientException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081。
        String endpoint = "localhost:8081";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        // 普通消息发送。
        Message message = provider.newMessageBuilder()
            .setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys("messageKey")
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag("messageTag")
            // 消息体。
            .setBody("messageBody".getBytes())
            .build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            logger.error("Failed to send message", e);
        }
        // producer.close();
    }
}

5.2.5 编写消费者

在已创建的Java工程中，创建订阅普通消息程序并运行。Apache RocketMQ 支持SimpleConsumer和PushConsumer两种消费者类型，您可以选择以下任意一种方式订阅消息。

import java.io.IOException;
import java.util.Collections;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PushConsumerExample {
    private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoints = "localhost:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组，Group需要提前创建。
        String consumerGroup = "YourConsumerGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组。
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器。
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果。
                logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                return ConsumeResult.SUCCESS;
            })
            .build();
        Thread.sleep(Long.MAX_VALUE);
        // 如果不需要再使用 PushConsumer，可关闭该实例。
        // pushConsumer.close();
    }
}

6.安装可视化控制台

RocketMQ提供了UI管理工具，名为rocketmq-console。

项目地址：https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

下载源码,编译。

修改rocketmq-console\src\main\resources\application.properties配置文件

修改namesrvAddr的地址。

## 指定端口号和命名空间地址
# java -jar rocketmq-console-ng-1.0.1.jar --server.port=9877 --rocketmq.config.namesrvAddr=106.15.0.30:9876 &

目录CONTENT

在Linux中用源码安装RocketMQ