RocketMQ是一款分布式、队列模型的消息中间件。支持集群模型、负载均衡、水平扩展能力，亿级别的消息堆积能力，采用零拷贝的原理、顺序写盘、随机读。丰富的API使用，代码优秀，底层通信框架使用Netty NIO框架，NameServer代替Zookeper。强调集群无单点，可扩展，任意一点高可用，水平可扩展。消息失败重试机制，消息可查询。开源社区活跃、成熟度高。（经过双十一的考验）

官网地址：GitHub - apache/rocketmq: Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications. - GitHub - apache/rocketmq: Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.https://github.com/apache/rocketmq

2、核心概念模型

Producer：消息生产者，负责产生消息，一般有业务喜用负责产生消息。

Consumer：消息消费者，负责消费消息，一般都是后台系统负责异步消费。

Push Consumer：Consumer的一种，需要向Consumer对象注册监听。

Pull Consumer：Consumer的一种，需要主动请求Broker拉取消息。

Producer Group：生产者集合，一般用于发送一类消息。

Consumer Group：消费者集合，一般用于接受一类消息进行消费。

Broker：MQ消息服务。（中转角色，用于消息存储与生产消费转发）

3、RocketMQ-源码包下载与结构说明

下载地址：Releases · apache/rocketmq · GitHub

RocketMQ的源码包结构

rocketmq-broker 主要的业务逻辑，消息收发，主从同步，pagecache

rocketmq-client 客户端接口，比如生产者和消费者。

rocketmq- example 示例，比如生产者和消费者。

rocketmq- common 公用数据结构等等。

rocketmq- distribution 编译模块，编译输出等。

rocketmq- filter 进行Broker过滤的不感兴趣的消息传输，减小带宽压力。

rocketmq- logappender、rocketmq-logging日志相关。

rocketmq-namesrv Namesrv服务，用于服务协调。

rocketmq-openmessaging 对外提供服务。

rockermq-remoting 远程调用接口，封装Netty底层通信。

rocketmq-srvutil 提供一些公用的工具方法，比如解析命令行参数。

rocketmq-srvutil 消息存储。

rocketmq-test 测试模块

rocketmq-tools 管理工具。

4、RocketMQ-环境搭建（搭建一个实例）

4.1、Hosts添加信息

vi /etc/hosts

在hosts文件底部编写(双主双从的话，两台机器配置文件一致，主节点、从节点)

电脑ip rocketmq-nameserver1
电脑ip rocketmq-master1

4.2、上传解压

在/usr/local 文件夹下创建文件夹apache-rocketmq，并将rockermq文件放入该文件夹。

mkdir /usr/local/apache-rocketmq

建立软连接

ln -s apache-rocketmq rocketmq

4.3、创建存储路径

mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index

4.4、编辑RocketMQ配置文件

vim /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties

brokerClusterName=rocketmq-cluster
#broker 名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer 地址，分号分割 
namesrvAddr=rocketmq-nameserver1:9876
#在发送消息时，自动创建服务器不存在的 topic，默认创建的队列数 
defaultTopicQueueNums=4
#是否允许 Broker 自动创建 Topic，建议线下开启，线上关闭 
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭 
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4 点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog 每个文件的大小默认 1G 
mapedFileSizeCommitLog=1073741824
#ConsumeQueue 每个文件默认存 30W 条，根据业务情况调整 
mapedFileSizeConsumeQueue=300000 
#destroyMapedFileIntervalForcibly=120000 
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径 
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径 
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径 
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4 
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000 
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制 Master 
#- SYNC_MASTER 同步双写 Master #- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘 
#- SYNC_FLUSH 同步刷盘 
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false 
#发消息线程池数量 
#sendMessageThreadPoolNums=128 
#拉消息线程池数量 
#pullMessageThreadPoolNums=128
#多网卡情况下需要配置brokerIP,否则broker启动后IP异常。两主两丛配置两个，其他的依次类推
brokerIP1=电脑IP
brokerIP2=电脑IP

4.5、修改日志配置文件

mkdir -p /usr/local/rocketmq/logs
cd /usr/local/rocketmq/conf && sed -i 's#$user.home#/usr/local/rocketmq#g' *.xml

4.6、修改启动脚本参数（默认分配的内存很大，放小一点）

# vim /usr/local/rocketmq/bin/runbroker.sh

修改JAVA_OPT启动参数：

JAVA_OPT="$JAVA_OPT -server -Xms1g -Xmx1g -Xmn512m -XX:PermSize=128m - XX:MaxPermSize=320m"

# vim /usr/local/rocketmq/bin/runserver.sh

修改JAVA_OPT启动参数

JAVA_OPT="$JAVA_OPT -server -Xms1g -Xmx1g -Xmn512m -XX:PermSize=128m - XX:MaxPermSize=320m"

4.7、启动 NameServer

# cd /usr/local/rocketmq/bin
# nohup sh mqnamesrv &

4.8、启动 BrokerServer

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties  >/dev/null 2>&1 &

4.9、查看是否启动成功

# jps
# tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
# tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log

4.10、数据清理

# cd /usr/local/rocketmq/bin
# sh mqshutdown broker
# sh mqshutdown namesrv
# --等待停止
# rm -rf /usr/local/rocketmq/store
# mkdir /usr/local/rocketmq/store
# mkdir /usr/local/rocketmq/store/commitlog
# mkdir /usr/local/rocketmq/store/consumequeue
# mkdir /usr/local/rocketmq/store/index
# --按照上面步骤重启 NameServer 与 BrokerServer

5、RocketMQ控制台部署与使用(messageId查询报错解决方法)

5.1、RocketMQ-Console下载地址

GitHub - SummerUnfair/rocketmq-externals: Mirror of Apache RocketMQ (Incubating)Mirror of Apache RocketMQ (Incubating). Contribute to SummerUnfair/rocketmq-externals development by creating an account on GitHub.https://github.com/SummerUnfair/rocketmq-externals

5.2、修改配置文件

修改配置文件中的rocketmq.config.namesrvAddr属性,该为自己RocketMQ的地址。其他的按需修改

5.3、当messageId查询报错的时候解决方案

修改源码MessageView类，修改fromMessageExt方法

5.4、修改后的RocketMQ-Console源码百度网盘地址

public static MessageView fromMessageExt(MessageExt messageExt) 
        MessageView messageView = new MessageView();
        BeanUtils.copyProperties(messageExt, messageView);
        if (messageExt.getBody() != null) 
            messageView.setMessageBody(new String(messageExt.getBody(), Charsets.UTF_8));
        
        if (messageExt instanceof MessageClientExt)
            MessageClientExt ext= (MessageClientExt) messageExt;
            messageView.setMsgId(ext.getOffsetMsgId());
        
        return messageView;

5.4、打包后，启动。

访问页面：http://localhost:8080

二、RocketMQ-急速入门

1、生产者模型使用

public class Producer 
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException 
        //创建DefaultMQProducer对象生产消息
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_producer_name");
        //指定RocketMQ Service 地址
        producer.setNamesrvAddr(Const.NAMESRV_ADDR);
        //启动RocketMQ
        producer.start();
        //循环五次发送消息
        for (int i = 0; i < 5; i++) 
             /*
                 发送消息
                 参数   topic 消息主题
                 参数   tags  消息标签
                 参数   keys  用户自定义的key,唯一标识
                 参数   body  消息体
            */
            Message testMessage = new Message("test_quick_topic",
                    "TagA",
                    "Key" + i,
                    ("Hello RocketMQ" + i).getBytes());
            SendResult sendResult = producer.send(testMessage);
            System.out.println("消息发出：" + sendResult);

        
        producer.shutdown();

2、消费者模型使用

public class Consumer 
    public static void main(String[] args) throws MQClientException 
        //创建DefaultMQPushConsumer对象消费消息
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
        //指定RocketMQ Service 地址
        consumer.setNamesrvAddr(Const.NAMESRV_ADDR);
        //设置消费方式，从最后一条消息消费，枚举可根据业务自行选择
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /*
           注册订阅
           topic 消息主题
           subExpression 参数支持通配符，值为*的话，消息主题(test_quick_topic)下的都会消费，也可以为具体的tags
         */
        consumer.subscribe("test_quick_topic", "TagA");
        //    监听topic,消费消息
        consumer.registerMessageListener(new MessageListenerConcurrently() 
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) 
                MessageExt messageExt = list.get(0);
                try 
                    String topic = messageExt.getTopic();
                    String tags = messageExt.getTags();
                    String keys = messageExt.getKeys();
                    if (keys.equals("Key1"))
                        System.out.println("消息消费失败");
                        int a=1/0;
                    
                    String msgBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    System.out.println("topic：" + topic + ",tags：" + tags + ",keys：" + keys + ",msgBody：" + msgBody);

                 catch (Exception e) 
                    e.printStackTrace();
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    System.out.println("reconsumeTimes ："+reconsumeTimes);
                    if (reconsumeTimes==3) 
                        //记录消费失败操作日志
                        System.out.println("记录消费失败操作日志");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            
        );
        consumer.start();
        System.out.println("consumer start ...");

3、四种集群环境

3.1、图例

3.2、四种模式概览

3.2.1、单点模式

3.2.2、主从模式

3.2.3、双主模式

3.2.4、双主双从模式，多主多从模式。

4、主从集群环境构建(双主双从)

链接: https://pan.baidu.com/s/1qhBgCENdEaBEDvtQtSFXhA?pwd=ibpi 提取码: ibpi
--来自百度网盘超级会员v4的分享

三、RocketMQ-生产者核心

1、配置参数解析

1.1、productGroup：组名。

1.2、createTopicKey：Topic Key，很少指定。

1.3、defaultTopicQueueNums：默认Topic队列数(默认为4)。

1.4、sendMsgTimeout：发送消息的超时时间(单位：ms)。

1.5、compressMsgBodyOverHowmuch：默认压缩字节4096。

1.6、retryTimesWhenSendFailed：当发送消息超时或者失败的时候，重发机制配置（有同步、异步重发，可配置）。

1.7、retryAnotherBrokerWhenNotStoreOK：没有存储成功，选择别的broker存储（默认false）。

1.8、maxMessageSize：消息体最大限制，默认为128K。

2、主从同步机制

2.1、Master - Slave 主从同步

2.2、同步信息：数据内容（实时同步,socket实现）+元数据信息（定时任务实现同步,Netty）

2.3、元数据同步：Broker角色识别，为Slave则启动同步任务

2.4、消息同步：HAService、HAconnection、WaitNotfiyObject。

3、同步/异步消息发送解析

3.1、消息同步发送：producer.send(msg)

3.2、同步发送消息核心实现：DefaultMQProducerImpl

3.3、消息异步发送：producer.send(Message msg,SendCallback sendCallback)

3.4、异步发送消息核心实现：DefaultMQProducerImpl

//异步发送消息
            producer.send(testMessage, new SendCallback() 
                //消息发送成功会掉
                @Override
                public void onSuccess(SendResult sendResult) 
                    System.out.println("msgId:" + sendResult.getMsgId() + ",status:" + sendResult.getSendStatus() + "消息发送成功");
                

                //消息发送失败会掉
                @Override
                public void onException(Throwable throwable) 
                    System.out.println("消息发送失败");
                
            );

4、延迟消息（Level可修改）

延迟消息：消息发到Broker后，要特定的时间才会被Consumer消费。

只支持固定精度的定时消息

MessageStoreConfig配置类&ScheduleMessageService任务类

setDelayTimeLevel方法设置

 Message testMessage = new Message("test_quick_topic",
                    "TagA",
                    "Key" + i,
                    ("Hello RocketMQ" + i).getBytes());
            if (i==1) 
                testMessage.setDelayTimeLevel(2);
            
            //同步发送消息
            SendResult sendResult = producer.send(testMessage);
            System.out.println("消息发出：" + sendResult);

5、消息的返回状态

5.1、SEND_OK：消息发送成功。

5.2、FLUSH_DISK_TIMEOUT：消息发送成功，但服务器刷盘的时候超时了

5.3、FLUSH_SLAVE_TIMEOUT：消息发送成功，从节点同步过程中超时了。

5.4、SLAVE_NOT_AVAILABLE：从节点不可用。

6、自定义消息发送规则（如何把消息发送到指定的队列）

producer.send(Msg,selector,Obj);

//指定队列发送消息(默认是的四个队列)
            SendResult send = producer.send(testMessage, new MessageQueueSelector() 
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) 
                    Integer queueNumber = (Integer) o;
                    return list.get(queueNumber);
                
            , 2);
            System.out.println(send);

四、RocketMQ-消费者核心

1、配置参数详解

1.1、consumeFromWhere：指定启动后broker开始消费位置

1.2、allocateMessageQueueStrategy：消息分配的策略（集群模式下使用）

1.3、subscription：标识订阅，第一个参数为订阅的主题，第二个参数是实现消息Tags过滤表达式（可以写具体的tags、*(不过滤)、tagsA||tagsB(包含过滤tagsA和tagsB)）

1.4、offsetStore：存储实际的偏移量，有本地还远程的方式

1.5、consumeThreadMin/consumeThreadMax：消费者线程池的自动调整配置

1.6、consumeConcurrentlyMaxSpan/pullThresholdForQueue：用来做流控的，第一个方法表示单个队列并行消费的最大跨度，第二个方法表示单个队列最大的消费个数是多少。

1.7、pullinterval/pullBatchSize：用作消息拉取，第一个方法表示消息拉取的时间间隔，第二个方法表示一次拉取的数据是多少

1.8、consumeMessageBatchMaxSize：默认为1，表示一次消息最多可以拉取多少条数据

2、集群与广播模式

2.1、集群模式

clustering模式默认为集群模式，GroupName用于把多个Consumer组织到一起，相同的GroupName的consumer只消费所订阅消息的一部分。目的是达到天然的负载均衡机制。

2.2、广播模式

Broadcasting模式（广播模式），同一个ConsumerGroup里的Consumer都会消费订阅Topic全部信息，也就是一条消息会被每一个Consumer消费，使用setMessageModel方法设置模式

3、消息存储核心-OffSet

Offset是消息消费进度的核心，Offset指某个topic下的一条消息在某个messageQueue里的位置。通过Offset可以进行定位到这条消息，Offset的存储实现分为远程文件类型和本地文件类型两种。集群模式（RemoteBrokerOffsetStore）采用远程文件存储offset，本质上因为多消费模式，每个Consumer消费所订阅主题的一部分。这种情况需要broker控制offset的值，使用RemoteBrokerOffsetStore。广播模式（LocalFileOffsetStore）由于每个Consumer都会收到消息且消费，各个Consumer之间没有任何干扰，独立线程消费，所以使用LocalFileOffsetStore,也就是把Offset存储到本地。

4、PullConsumer

消息拉取方式：DefaultMQPullConsumer,Pull方式主要做了三件事，获取Message Queue并遍历，维护OffsetStore，根据不同的消息状态做不同的处理。

五、分布式事务消息讲解

1、分布式事务消息理解

2、RocketMQ事务消息机制分析

3、支付场景去重机制讲解

4、支付场景下分布式事务实战

以上是关于Rocket详细教程的主要内容，如果未能解决你的问题，请参考以下文章

java高并发框架 SSM框架详细整合教程(Spring+SpringMVC+MyBatis)

Spring入门详细教程

spring boot教程网盘下载

Spring入门详细教程

spring-boo hello world程序