关键词：
目录
5、RocketMQ控制台部署与使用(messageId查询报错解决方法)
5.4、修改后的RocketMQ-Console源码百度网盘地址
一、RocketMQ-初探门径
1、RocketMQ整体介绍
RocketMQ是一款分布式、队列模型的消息中间件。支持集群模型、负载均衡、水平扩展能力，亿级别的消息堆积能力，采用零拷贝的原理、顺序写盘、随机读。丰富的API使用，代码优秀，底层通信框架使用Netty NIO框架，NameServer代替Zookeper。强调集群无单点，可扩展，任意一点高可用，水平可扩展。消息失败重试机制，消息可查询。开源社区活跃、成熟度高。（经过双十一的考验）
https://github.com/apache/rocketmq2、核心概念模型
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-master14.2、上传解压
在/usr/local 文件夹下创建文件夹apache-rocketmq，并将rockermq文件放入该文件夹。
mkdir /usr/local/apache-rocketmq建立软连接
ln -s apache-rocketmq rocketmq4.3、创建存储路径
mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index4.4、编辑RocketMQ配置文件
vim /usr/local/rocketmq/conf/2m-2s-async/broker-a.propertiesbrokerClusterName=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' *.xml4.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.log4.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 与 BrokerServer5、RocketMQ控制台部署与使用(messageId查询报错解决方法)
5.1、RocketMQ-Console下载地址
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、支付场景下分布式事务实战
超详细pytorch入门教程(代码片段)
...其他操作Pytorch入门教程（一）请参考：【超详细】Pytorch入门教程（一）学 查看详情
codeforcesround#499(div.2)d.rocket(代码片段)
题目连接：D.Rocket题意：交互题，让你猜一个数字，你每次询问一个数字，他会告诉你是大还是小，但是他回答不一定正确，有一个序列p,当p[i]为1时他会回答正确的，否则错误。询问次数<60,题解：很明显的二分，但是p的长度... 查看详情
ffmpeg教程（超级详细版）(代码片段)
一、参考资料通过ffmpeg把图片转换成视频FFmpeg命令(一)、使用filter_complex命令拼接视频FFmpeg视频处理入门教程给新手的20多个FFmpeg命令示例FFmpeg命令行转码ffmpeg翻译文档(ffmpeg-all包含重要组件）FFmpegFiltersDocumentationFFmpeg命令行滤... 查看详情
windows安装rabbitmq详细教程(代码片段)
Windows安装RabbitMQ详细教程前言一、RabbitMQ是什么？二、安装步骤1.安装准备工作2.开始安装3.结束安装总结前言下面我将分享下RabbitMQ最新版本在Windows上的详细安装教程。笔者在这里使用的系统环境如下：windows系统：Win... 查看详情
jenkins详细安装教程(代码片段)
1.先下载msi文件 Jenkins下载链接:https://pan.baidu.com/s/1SACKNgW7OZrJoXMRDhsJxQ提取码:94b9 2.安装 解压后得到的是jenkins-2.9.1.msi文件，双击运行并安装即可。 3.进入jenkins： 安装成功后在... 查看详情
nvjdc安装详细教程。(代码片段)
首先感谢原作者NolanHzy分享，仓库地址https://github.com/NolanHzy/nvjdc开始进入主题.安装nvjdc1.安装解压程序(已安装的略过)yuminstallwgetunzip-y2.创建一个目录放配置chromium资源mkdirnolanjdc&&cdnolanjdc3.下载config.json配置文件并且修改... 查看详情
numpy超详细教程：ndarray的内部机理及高级迭代(代码片段)
系列文章地址NumPy最详细教程（1）：NumPy数组NumPy超详细教程（2）：数据类型NumPy超详细教程（3）：ndarray的内部机理及高级迭代ndarray对象的内部机理在前面的内容中，我们已经详细讲述了ndarray的使用，在本章的开始部分，... 查看详情
nginx功能搭建详细教程(代码片段)
一、访问控制基于用户名、密码的验证访问服务端配置vim/usr/local/nginx/conf/nginx.conf在想验证的location下面添加,以根区域为例location/ roothtml; indexindex.htmlindex.htm; #添加下面两行 auth_basic"welcomeyouhere"; auth_basic_user_file/ 查看详情
intellijidea详细配置和使用教程（转）(代码片段)
http://blog.csdn.net/m_m254282520/article/details/78900238 关闭IntellijIDEA自动更新在File->Settings->Appearance&Behavior->SystemSettings->Updates下取消Automaticallycheckupdatesfor勾选&n 查看详情
vmware安装linux详细教程(代码片段)
...nux与磁盘分区、配置网络、设置快照、克隆系统四部分的详细教程。一、VMware安装Linux除基本自定义安装外，还包含对磁盘的分区1.点击新建虚拟机，选择自定义，典型的较为简单，但是我们选择自定义安装2.点击【下一步】选择... 查看详情
hadoop单点安装详细教程(代码片段)
系统：Ubuntu18.04Hadoop：3.3.0系统环境准备sudoapt-getupdatesudoapt-getinstalldefault-jdksudoapt-getinstallvimsudoapt-getinstallssh创建一个用户，命名为hadoop（或者其他的也可）sudouseraddhadoop之后输 查看详情
mysql主从同步详细教程(代码片段)
例子： 假如我需要同步test1、test2数据库 系统：centos7 主库主机：192.168.1.252 从库主机：192.168.1.251 端口都是：3306 (1)、打开主数据的配置文件my.cnf，添加这几项（2）、重启mysql（service... 查看详情
mysql主从同步详细教程(代码片段)
例子： 假如我需要同步test1、test2数据库 系统：centos7 主库主机：192.168.1.252 从库主机：192.168.1.251 端口都是：3306 (1)、打开主数据的配置文件my.cnf，添加这几项（2）、重启mysql（service... 查看详情
mysql主从同步详细教程(代码片段)
例子： 假如我需要同步test1、test2数据库 系统：centos7 主库主机：192.168.1.252 从库主机：192.168.1.251 端口都是：3306 (1)、打开主数据的配置文件my.cnf，添加这几项（2）、重启mysql（service... 查看详情
json11详细使用教程，持续更新！(代码片段)
json11详细使用教程，持续更新jsonjson11基本使用一些json11的具体操作json11解析数组，并基于范围循环遍历参考资料jsonJSON的全称是”JavaScriptObjectNotation”，意思是JavaScript对象表示法，它是一种基于文本，独立于... 查看详情
java实现2048游戏之详细教程(代码片段)
一、整体项目结构使用Maven来管理项目结构 二、基本功能实现（一）创建游戏窗口（静态）packagecom.baidu.czy;importjava.applet.Applet;importjava.applet.AudioClip;importjava.awt.*;importjava.awt.event.ActionEvent;importjav 查看详情
visualstudio2017安装使用教程（详细版）(代码片段)
VisualStudio2017安装使用教程（详细） 本人曾因无法使用vs编写C语言程序痛苦一个月之久，实乃惭愧，后发现不少同学也同样存在着相同问题，其原因归结于网上的各种教程没有对特殊情况进行讲解... 查看详情
codepush热更新组件详细接入教程(代码片段)
CodePush热更新组件详细接入教程什么是CodePushCodePush是一个微软开发的云服务器。通过它，开发者可以直接在用户的设备上部署手机应用更新。CodePush相当于一个中心仓库，开发者可以推送当前的更新（包括JS/HTML/CSS/IMAGE等）到CoduP... 查看详情