1. 环境准备

Linux
JDK 1.8
至少 4GB 内存

2 下载 RocketMQ

# 方式1：下载二进制包（推荐）
wget https://archive.apache.org/dist/rocketmq/4.9.4/rocketmq-all-4.9.4-bin-release.zip
unzip rocketmq-all-4.9.4-bin-release.zip
cd rocketmq-all-4.9.4-bin-release

方式2下载源码，需要自己构建，还需要依赖构建工具（如maven，不推荐）

3 单节点部署

（1）. 调整内存配置

# 进入 RocketMQ 目录
cd rocketmq-all-4.9.4-bin-release

# 修改 NameServer 内存配置
vim bin/runserver.sh
# 找到 JAVA_OPT，修改为（根据机器内存调整）
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"

# 修改 Broker 内存配置
vim bin/runbroker.sh
# 修改为
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m"

（2）启动nameserver

# 启动 NameServer（后台运行）
nohup sh bin/mqnamesrv &

# 查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log
# 看到 "The Name Server boot success." 表示启动成功

# 或者前台启动（调试用）
sh bin/mqnamesrv

（3）配置broker

# 创建 Broker 配置文件
vim conf/broker.conf
# Broker 配置
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0

# NameServer 地址
namesrvAddr = localhost:9876

# 存储路径
storePathRootDir = /tmp/rocketmq/store
storePathCommitLog = /tmp/rocketmq/store/commitlog
storePathConsumeQueue = /tmp/rocketmq/store/consumequeue
storePathIndex = /tmp/rocketmq/store/index

# 消息存储大小（根据磁盘调整）
mapedFileSizeCommitLog = 1073741824  # 1GB

# 删除过期文件时间（小时，默认72小时）
fileReservedTime = 72

# Broker 角色
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH

# 自动创建主题（开发环境开启）
autoCreateTopicEnable = true

（4）启动broker

# 使用自定义配置文件启动 Broker
nohup sh bin/mqbroker -c conf/broker.conf &

# 查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log
# 看到 "The broker[broker-a, 192.168.x.x:10911] boot success." 表示成功

4 验证安装成功与否

# 查看进程
jps
# 应看到 NamesrvStartup 和 BrokerStartup

# 检查端口监听
netstat -tlnp | grep -E '9876|10911|10909'
# 9876: NameServer
# 10911: Broker 服务端口
# 10909: Broker VIP 通道

# 使用管理命令查看状态
sh bin/mqadmin clusterList -n localhost:9876

5 快速测试

（1）发送消息

# 设置环境变量（临时）
export NAMESRV_ADDR=localhost:9876

# 发送测试消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

（2）消费消息

# 消费消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

消费报错了。

（原因是因为我另开了一个服务器窗口，未设置环境变量）...

export NAMESRV_ADDR=localhost:9876 后再执行Consumer即可

6 java程序

pom依赖

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

消息发送

（1）创建消息生产者

public static DefaultMQProducer getDefaultMQProducer() {
        if(producer == null){
            synchronized (MyProducer.class) {
                // 1.初始化producer
                producer = new DefaultMQProducer(CFG.getString(ConfigConstants.ROCKETMQ_PRODUCERGROUP));
                // 2.设置NameServer的地址
                producer.setNamesrvAddr(CFG.getString(ConfigConstants.ROCKETMQ_ONSADDR));
                // 3.启动生产者实例
                try {
                    producer.start();
                } catch (MQClientException e) {
                    logger.error("启动生产者失败", e);
                    throw new BizException(GlobalExcpEnum.SELF_MQ_ERROR);
                }
                logger.info("SelfMq DefaultMQProducer start success!");
            }
        }
        return producer;
    }

注意：这里的nameserver地址加到了本地配置文件里， 如果是单点部署，就说 ip:9876，注意不要带http://

(2) 发送消息

 public static void sendMsgAsync(String topic, final MqMessage msg) {
        //topic 从配置中获取
        String topicEnv = cfg.getString(ConfigConstants.ROCKETMQ_TOPIC_ENV);
        if (StringUtils.isBlank(topicEnv)) {
            throw new BizException(GlobalExcpEnum.TOPIC_ENV_EMPTY);
        }
        topic = topic + topicEnv;
        //初始化消息   MqMessage为业务自定义的消息实体，其中body就是消息体
        Message msgs = new Message(topic, msg.getTag(), msg.getKey(), msg.getBody());
        long delayTime = msg.getDelayTime();
        if (delayTime > 0){
            long millis = System.currentTimeMillis();
            log.debug("设置延迟消息, 当前时间:{}, 延迟毫秒:{}",millis,delayTime);
            // RocketMQ 延迟级别对照表
            //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            // 对应关系：
            // level 1 = 1秒
            // level 2 = 5秒
            // level 3 = 10秒
            // level 4 = 30秒
            // level 5 = 1分钟
            // ... 以此类推
            // level 18 = 2小时

            msgs.setDelayTimeLevel(2);
        }
        log.debug("printMqSendMsg, topicName:{}, key:{}, tag:{} , body:{}",topic,msg.getKey(),msg.getTag(),msg.getBody() == null ? "" : new String(msg.getBody()));

        if (producer == null) {
            log.info("MQ 没初始化, topic:{}", topic);
            producer = MyProducer.getDefaultMQProducer();
        }

        try {
            producer.send(msgs, new org.apache.rocketmq.client.producer.SendCallback() {
                @Override
                public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
                    log.debug("send message success. topic:{}, msgId:{}, tag:{}, key:{}", sendResult.getRegionId(),
                            sendResult.getMsgId(), msg.getTag(), msg.getKey());
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("send message failed. topic:{},  tag:{}, key:{} ", msgs.getTopic(), msg.getTag(), msg.getKey());
                    log.error("mq send exception", throwable);
                }
            });
        } catch (MQClientException e) {
           log.error("mq", e);
        } catch (RemotingException e) {
            log.error("mq", e);
        } catch (InterruptedException e) {
            log.error("inter", e);
        }
    }

消息消费

（1）初始化消费者

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("指定消费groupid");

//设置rocketmq的nameserver地址

//设置最大重试次数

//设置消费线程数

（2）订阅topic&tag

（3）注册消息监听者

完整自定义消费者代码如下

public class MyConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MyConsumer.class);
    //并发消费   还有一个顺序消费类型，这里根据业务需求选择不同的消费类型
    private MessageListenerConcurrently listener;
    private static DefaultMQPushConsumer consumer;


    public void init(InitInfo initInfo){
        consumer = new DefaultMQPushConsumer(initInfo.getGroupID());
        consumer.setNamesrvAddr(initInfo.getoNSAddr());
        consumer.setMaxReconsumeTimes(initInfo.getMaxReconsumeTimes());
        consumer.setConsumeThreadMax(initInfo.getConsumeThreadNums());
        consumer.setConsumeThreadMin(initInfo.getConsumeThreadNums());

        logger.debug("initInfo:{}", JsonUtils.toJSONString(initInfo));

        String[] topicTags = initInfo.getTopictag().split(",");
        for (String topicTag : topicTags) {
            String[] s = topicTag.split(":");
            try {
                consumer.subscribe(s[0], s[1]);
            } catch (MQClientException e) {
                logger.error("消费者订阅异常", e);
                throw new BizException(GlobalExcpEnum.SELF_MQ_ERROR);
            }
        }
        consumer.registerMessageListener(listener);
        try {
            consumer.start();
        } catch (MQClientException e) {
            logger.error("消费者启动异常", e);
            throw new BizException(GlobalExcpEnum.SELF_MQ_ERROR);
        }
        logger.info("###### MQ consumer started, topiocTags - {} ######", initInfo.getTopictag());

        Thread shutdownHookOne = new Thread() {
            @Override
            public void run() {
                MyConsumer.destroy();
            }
        };
        Runtime.getRuntime().addShutdownHook(shutdownHookOne);
    }


    public static void destroy() {
        logger.info("rocketmq Consumer shutdown!");
        if(consumer!=null){
            consumer.shutdown();
            logger.info("rocketmq Consumer shutdown success!");
        }
    }

    public MessageListenerConcurrently getListener() {
        return listener;
    }

    public void setListener(MessageListenerConcurrently listener) {
        this.listener = listener;
    }

其他-RocketMQ 延迟级别对照表

        private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

// 对应关系：
// level 1 = 1秒
// level 2 = 5秒
// level 3 = 10秒
// level 4 = 30秒
// level 5 = 1分钟
// ... 以此类推
// level 18 = 2小时

阿里云产品的mq客户端已经支持设置延迟时间。