Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
码出高效码出未来
Rocketmq
提交
b9c108b6
R
Rocketmq
项目概览
码出高效码出未来
/
Rocketmq
与 Fork 源项目一致
Fork自
Apache RocketMQ / Rocketmq
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
Rocketmq
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
b9c108b6
编写于
6月 16, 2021
作者:
Z
zongtanghu
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/develop_include_3_commits' into develop
上级
df1d93fc
57c166bc
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
13 addition
and
6 deletion
+13
-6
docs/cn/design.md
docs/cn/design.md
+1
-1
tools/src/main/java/org/apache/rocketmq/tools/command/message/SendMessageCommand.java
...he/rocketmq/tools/command/message/SendMessageCommand.java
+12
-5
未找到文件。
docs/cn/design.md
浏览文件 @
b9c108b6
...
...
@@ -10,7 +10,7 @@
#### 1.1 消息存储整体架构
消息存储架构图中主要有下面三个跟消息存储相关的文件构成。
(1) CommitLog:消息主体以及元数据的存储主体,存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G
,
文件名长度为20位,左边补零,剩余为起始偏移量,比如00000000000000000000代表了第一个文件,起始偏移量为0,文件大小为1G=1073741824;当第一个文件写满了,第二个文件为00000000001073741824,起始偏移量为1073741824,以此类推。消息主要是顺序写入日志文件,当文件满了,写入下一个文件;
(1) CommitLog:消息主体以及元数据的存储主体,存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G
,
文件名长度为20位,左边补零,剩余为起始偏移量,比如00000000000000000000代表了第一个文件,起始偏移量为0,文件大小为1G=1073741824;当第一个文件写满了,第二个文件为00000000001073741824,起始偏移量为1073741824,以此类推。消息主要是顺序写入日志文件,当文件满了,写入下一个文件;
(2) ConsumeQueue:消息消费队列,引入的目的主要是提高消息消费的性能,由于RocketMQ是基于主题topic的订阅模式,消息消费是针对主题进行的,如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中,ConsumeQueue(逻辑消费队列)作为消费消息的索引,保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset,消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件,故consumequeue文件夹的组织方式如下:topic/queue/file三层组织结构,具体存储路径为:$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计,每一个条目共20个字节,分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode,单个文件由30W个条目组成,可以像数组一样随机访问每一个条目,每个ConsumeQueue文件大小约5.72M;
...
...
tools/src/main/java/org/apache/rocketmq/tools/command/message/SendMessageCommand.java
浏览文件 @
b9c108b6
...
...
@@ -68,14 +68,18 @@ public class SendMessageCommand implements SubCommand {
opt
.
setRequired
(
false
);
options
.
addOption
(
opt
);
opt
=
new
Option
(
"m"
,
"msgTraceEnable"
,
true
,
"Message Trace Enable, Default: false"
);
opt
.
setRequired
(
false
);
options
.
addOption
(
opt
);
return
options
;
}
private
DefaultMQProducer
createProducer
(
RPCHook
rpcHook
)
{
private
DefaultMQProducer
createProducer
(
RPCHook
rpcHook
,
boolean
msgTraceEnable
)
{
if
(
this
.
producer
!=
null
)
{
return
producer
;
}
else
{
producer
=
new
DefaultMQProducer
(
rpcHook
);
producer
=
new
DefaultMQProducer
(
null
,
rpcHook
,
msgTraceEnable
,
null
);
producer
.
setProducerGroup
(
Long
.
toString
(
System
.
currentTimeMillis
()));
return
producer
;
}
...
...
@@ -112,8 +116,11 @@ public class SendMessageCommand implements SubCommand {
}
catch
(
Exception
e
)
{
throw
new
RuntimeException
(
this
.
getClass
().
getSimpleName
()
+
" command failed"
,
e
);
}
DefaultMQProducer
producer
=
this
.
createProducer
(
rpcHook
);
boolean
msgTraceEnable
=
false
;
if
(
commandLine
.
hasOption
(
'm'
))
{
msgTraceEnable
=
Boolean
.
parseBoolean
(
commandLine
.
getOptionValue
(
'm'
).
trim
());
}
DefaultMQProducer
producer
=
this
.
createProducer
(
rpcHook
,
msgTraceEnable
);
SendResult
result
;
try
{
producer
.
start
();
...
...
@@ -153,4 +160,4 @@ public class SendMessageCommand implements SubCommand {
);
}
}
}
\ No newline at end of file
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录