Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
逝缘~
Rocketmq
提交
9422bdb3
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,发现更多精彩内容 >>
提交
9422bdb3
编写于
6月 02, 2017
作者:
王
王松松
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
correct spelling mistake
上级
bfa7f1ac
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
47 addition
and
43 deletion
+47
-43
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
...ient/impl/consumer/ConsumeMessageConcurrentlyService.java
+15
-14
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
...mq/client/impl/consumer/ConsumeMessageOrderlyService.java
+12
-11
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+15
-14
store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
...src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+5
-4
未找到文件。
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
浏览文件 @
9422bdb3
...
...
@@ -16,19 +16,6 @@
*/
package
org.apache.rocketmq.client.impl.consumer
;
import
java.util.ArrayList
;
import
java.util.Collections
;
import
java.util.HashMap
;
import
java.util.Iterator
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.concurrent.BlockingQueue
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.LinkedBlockingQueue
;
import
java.util.concurrent.RejectedExecutionException
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.ThreadPoolExecutor
;
import
java.util.concurrent.TimeUnit
;
import
org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
;
...
...
@@ -48,6 +35,20 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import
org.apache.rocketmq.remoting.common.RemotingHelper
;
import
org.slf4j.Logger
;
import
java.util.ArrayList
;
import
java.util.Collections
;
import
java.util.HashMap
;
import
java.util.Iterator
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.concurrent.BlockingQueue
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.LinkedBlockingQueue
;
import
java.util.concurrent.RejectedExecutionException
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.ThreadPoolExecutor
;
import
java.util.concurrent.TimeUnit
;
public
class
ConsumeMessageConcurrentlyService
implements
ConsumeMessageService
{
private
static
final
Logger
log
=
ClientLogger
.
getLog
();
private
final
DefaultMQPushConsumerImpl
defaultMQPushConsumerImpl
;
...
...
@@ -163,7 +164,7 @@ public class ConsumeMessageConcurrentlyService implements ConsumeMessageService
final
long
beginTime
=
System
.
currentTimeMillis
();
log
.
info
(
"consumeMessageDirectly receive new messge: {}"
,
msg
);
log
.
info
(
"consumeMessageDirectly receive new mess
a
ge: {}"
,
msg
);
try
{
ConsumeConcurrentlyStatus
status
=
this
.
messageListener
.
consumeMessage
(
msgs
,
context
);
...
...
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
浏览文件 @
9422bdb3
...
...
@@ -16,16 +16,6 @@
*/
package
org.apache.rocketmq.client.impl.consumer
;
import
java.util.ArrayList
;
import
java.util.Collections
;
import
java.util.HashMap
;
import
java.util.List
;
import
java.util.concurrent.BlockingQueue
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.LinkedBlockingQueue
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.ThreadPoolExecutor
;
import
java.util.concurrent.TimeUnit
;
import
org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus
;
...
...
@@ -48,6 +38,17 @@ import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import
org.apache.rocketmq.remoting.common.RemotingHelper
;
import
org.slf4j.Logger
;
import
java.util.ArrayList
;
import
java.util.Collections
;
import
java.util.HashMap
;
import
java.util.List
;
import
java.util.concurrent.BlockingQueue
;
import
java.util.concurrent.Executors
;
import
java.util.concurrent.LinkedBlockingQueue
;
import
java.util.concurrent.ScheduledExecutorService
;
import
java.util.concurrent.ThreadPoolExecutor
;
import
java.util.concurrent.TimeUnit
;
public
class
ConsumeMessageOrderlyService
implements
ConsumeMessageService
{
private
static
final
Logger
log
=
ClientLogger
.
getLog
();
private
final
static
long
MAX_TIME_CONSUME_CONTINUOUSLY
=
...
...
@@ -143,7 +144,7 @@ public class ConsumeMessageOrderlyService implements ConsumeMessageService {
final
long
beginTime
=
System
.
currentTimeMillis
();
log
.
info
(
"consumeMessageDirectly receive new messge: {}"
,
msg
);
log
.
info
(
"consumeMessageDirectly receive new mess
a
ge: {}"
,
msg
);
try
{
ConsumeOrderlyStatus
status
=
this
.
messageListener
.
consumeMessage
(
msgs
,
context
);
...
...
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
浏览文件 @
9422bdb3
...
...
@@ -16,13 +16,6 @@
*/
package
org.apache.rocketmq.store
;
import
java.nio.ByteBuffer
;
import
java.util.ArrayList
;
import
java.util.HashMap
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.concurrent.CountDownLatch
;
import
java.util.concurrent.TimeUnit
;
import
org.apache.rocketmq.common.ServiceThread
;
import
org.apache.rocketmq.common.UtilAll
;
import
org.apache.rocketmq.common.constant.LoggerName
;
...
...
@@ -39,6 +32,14 @@ import org.apache.rocketmq.store.schedule.ScheduleMessageService;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.nio.ByteBuffer
;
import
java.util.ArrayList
;
import
java.util.HashMap
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.concurrent.CountDownLatch
;
import
java.util.concurrent.TimeUnit
;
/**
* Store all metadata downtime for recovery, data protection reliability
*/
...
...
@@ -183,7 +184,7 @@ public class CommitLog {
index
++;
if
(
index
>=
mappedFiles
.
size
())
{
// Current branch can not happen
log
.
info
(
"recover last 3 physics file over, last maped file "
+
mappedFile
.
getFileName
());
log
.
info
(
"recover last 3 physics file over, last map
p
ed file "
+
mappedFile
.
getFileName
());
break
;
}
else
{
mappedFile
=
mappedFiles
.
get
(
index
);
...
...
@@ -417,7 +418,7 @@ public class CommitLog {
for
(;
index
>=
0
;
index
--)
{
mappedFile
=
mappedFiles
.
get
(
index
);
if
(
this
.
isMappedFileMatchedRecover
(
mappedFile
))
{
log
.
info
(
"recover from this maped file "
+
mappedFile
.
getFileName
());
log
.
info
(
"recover from this map
p
ed file "
+
mappedFile
.
getFileName
());
break
;
}
}
...
...
@@ -459,7 +460,7 @@ public class CommitLog {
if
(
index
>=
mappedFiles
.
size
())
{
// The current branch under normal circumstances should
// not happen
log
.
info
(
"recover physics file over, last maped file "
+
mappedFile
.
getFileName
());
log
.
info
(
"recover physics file over, last map
p
ed file "
+
mappedFile
.
getFileName
());
break
;
}
else
{
mappedFile
=
mappedFiles
.
get
(
index
);
...
...
@@ -585,7 +586,7 @@ public class CommitLog {
mappedFile
=
this
.
mappedFileQueue
.
getLastMappedFile
(
0
);
// Mark: NewFile may be cause noise
}
if
(
null
==
mappedFile
)
{
log
.
error
(
"create maped file1 error, topic: "
+
msg
.
getTopic
()
+
" clientAddr: "
+
msg
.
getBornHostString
());
log
.
error
(
"create map
p
ed file1 error, topic: "
+
msg
.
getTopic
()
+
" clientAddr: "
+
msg
.
getBornHostString
());
beginTimeInLock
=
0
;
return
new
PutMessageResult
(
PutMessageStatus
.
CREATE_MAPEDFILE_FAILED
,
null
);
}
...
...
@@ -600,7 +601,7 @@ public class CommitLog {
mappedFile
=
this
.
mappedFileQueue
.
getLastMappedFile
(
0
);
if
(
null
==
mappedFile
)
{
// XXX: warn and notify me
log
.
error
(
"create maped file2 error, topic: "
+
msg
.
getTopic
()
+
" clientAddr: "
+
msg
.
getBornHostString
());
log
.
error
(
"create map
p
ed file2 error, topic: "
+
msg
.
getTopic
()
+
" clientAddr: "
+
msg
.
getBornHostString
());
beginTimeInLock
=
0
;
return
new
PutMessageResult
(
PutMessageStatus
.
CREATE_MAPEDFILE_FAILED
,
result
);
}
...
...
@@ -735,7 +736,7 @@ public class CommitLog {
mappedFile
=
this
.
mappedFileQueue
.
getLastMappedFile
(
0
);
// Mark: NewFile may be cause noise
}
if
(
null
==
mappedFile
)
{
log
.
error
(
"Create maped file1 error, topic: {} clientAddr: {}"
,
messageExtBatch
.
getTopic
(),
messageExtBatch
.
getBornHostString
());
log
.
error
(
"Create map
p
ed file1 error, topic: {} clientAddr: {}"
,
messageExtBatch
.
getTopic
(),
messageExtBatch
.
getBornHostString
());
beginTimeInLock
=
0
;
return
new
PutMessageResult
(
PutMessageStatus
.
CREATE_MAPEDFILE_FAILED
,
null
);
}
...
...
@@ -750,7 +751,7 @@ public class CommitLog {
mappedFile
=
this
.
mappedFileQueue
.
getLastMappedFile
(
0
);
if
(
null
==
mappedFile
)
{
// XXX: warn and notify me
log
.
error
(
"Create maped file2 error, topic: {} clientAddr: {}"
,
messageExtBatch
.
getTopic
(),
messageExtBatch
.
getBornHostString
());
log
.
error
(
"Create map
p
ed file2 error, topic: {} clientAddr: {}"
,
messageExtBatch
.
getTopic
(),
messageExtBatch
.
getBornHostString
());
beginTimeInLock
=
0
;
return
new
PutMessageResult
(
PutMessageStatus
.
CREATE_MAPEDFILE_FAILED
,
result
);
}
...
...
store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
浏览文件 @
9422bdb3
...
...
@@ -16,14 +16,15 @@
*/
package
org.apache.rocketmq.store
;
import
java.io.File
;
import
java.nio.ByteBuffer
;
import
java.util.List
;
import
org.apache.rocketmq.common.constant.LoggerName
;
import
org.apache.rocketmq.store.config.StorePathConfigHelper
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.io.File
;
import
java.nio.ByteBuffer
;
import
java.util.List
;
public
class
ConsumeQueue
{
public
static
final
int
CQ_STORE_UNIT_SIZE
=
20
;
...
...
@@ -121,7 +122,7 @@ public class ConsumeQueue {
index
++;
if
(
index
>=
mappedFiles
.
size
())
{
log
.
info
(
"recover last consume queue file over, last maped file "
log
.
info
(
"recover last consume queue file over, last map
p
ed file "
+
mappedFile
.
getFileName
());
break
;
}
else
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录