Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
逝缘~
Rocketmq
提交
a56bec81
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,发现更多精彩内容 >>
提交
a56bec81
编写于
11月 25, 2019
作者:
翊
翊名
提交者:
dinglei
11月 26, 2019
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feat(PullConsumer) support begin/end seek support for pull consumer
上级
c3b7d186
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
67 addition
and
16 deletion
+67
-16
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
...che/rocketmq/client/consumer/DefaultLitePullConsumer.java
+21
-7
client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java
...org/apache/rocketmq/client/consumer/LitePullConsumer.java
+23
-0
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
...tmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
+23
-9
未找到文件。
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
浏览文件 @
a56bec81
...
...
@@ -18,7 +18,6 @@ package org.apache.rocketmq.client.consumer;
import
java.util.Collection
;
import
java.util.List
;
import
org.apache.rocketmq.client.ClientConfig
;
import
org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely
;
import
org.apache.rocketmq.client.consumer.store.OffsetStore
;
...
...
@@ -35,12 +34,12 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
private
final
DefaultLitePullConsumerImpl
defaultLitePullConsumerImpl
;
/**
* Consumers belonging to the same consumer group share a group id. The consumers in a group then
*
divides the topic as fairly amongst themselves as possible by establishing that each queue is only
*
consumed by a single consumer from the group. If all consumers are from the same group, it functions
*
as a traditional message queue. Each message would be consumed by one consumer of the group only.
*
When multiple consumer groups exist, the flow of the data consumption model aligns with the traditiona
l
*
publish-subscribe model. The messages are broadcast to all
consumer groups.
* Consumers belonging to the same consumer group share a group id. The consumers in a group then
divides the topic
*
as fairly amongst themselves as possible by establishing that each queue is only consumed by a single consumer
*
from the group. If all consumers are from the same group, it functions as a traditional message queue. Each
*
message would be consumed by one consumer of the group only. When multiple consumer groups exist, the flow of the
*
data consumption model aligns with the traditional publish-subscribe model. The messages are broadcast to al
l
* consumer groups.
*/
private
String
consumerGroup
;
...
...
@@ -266,6 +265,21 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
return
this
.
defaultLitePullConsumerImpl
.
committed
(
messageQueue
);
}
@Override
public
void
updateNameServerAddress
(
String
nameServerAddress
)
{
this
.
defaultLitePullConsumerImpl
.
updateNameServerAddr
(
nameServerAddress
);
}
@Override
public
void
seekToBegin
(
MessageQueue
messageQueue
)
throws
MQClientException
{
this
.
defaultLitePullConsumerImpl
.
seekToBegin
(
messageQueue
);
}
@Override
public
void
seekToEnd
(
MessageQueue
messageQueue
)
throws
MQClientException
{
this
.
defaultLitePullConsumerImpl
.
seekToEnd
(
messageQueue
);
}
@Override
public
boolean
isAutoCommit
()
{
return
autoCommit
;
...
...
client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java
浏览文件 @
a56bec81
...
...
@@ -172,4 +172,27 @@ public interface LitePullConsumer {
*/
void
registerTopicMessageQueueChangeListener
(
String
topic
,
TopicMessageQueueChangeListener
topicMessageQueueChangeListener
)
throws
MQClientException
;
/**
* Update name server addresses.
*/
void
updateNameServerAddress
(
String
nameServerAddress
);
/**
* Overrides the fetch offsets with the begin offset that the consumer will use on the next poll. If this API is
* invoked for the same message queue more than once, the latest offset will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of consumption.
*
* @param messageQueue
*/
void
seekToBegin
(
MessageQueue
messageQueue
)
throws
MQClientException
;
/**
* Overrides the fetch offsets with the end offset that the consumer will use on the next poll. If this API is
* invoked for the same message queue more than once, the latest offset will be used on the next poll(). Note that
* you may lose data if this API is arbitrarily used in the middle of consumption.
*
* @param messageQueue
*/
void
seekToEnd
(
MessageQueue
messageQueue
)
throws
MQClientException
;
}
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
浏览文件 @
a56bec81
...
...
@@ -16,16 +16,16 @@
*/
package
org.apache.rocketmq.client.impl.consumer
;
import
java.util.ArrayList
;
import
java.util.Collection
;
import
java.util.Collections
;
import
java.util.HashMap
;
import
java.util.List
;
import
java.util.ArrayList
;
import
java.util.HashSet
;
import
java.util.Iterator
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.Set
;
import
java.util.Collection
;
import
java.util.HashSet
;
import
java.util.Properties
;
import
java.util.Set
;
import
java.util.concurrent.BlockingQueue
;
import
java.util.concurrent.ConcurrentHashMap
;
import
java.util.concurrent.ConcurrentMap
;
...
...
@@ -35,13 +35,12 @@ import java.util.concurrent.ScheduledExecutorService;
import
java.util.concurrent.ScheduledThreadPoolExecutor
;
import
java.util.concurrent.ThreadFactory
;
import
java.util.concurrent.TimeUnit
;
import
org.apache.rocketmq.client.Validators
;
import
org.apache.rocketmq.client.consumer.DefaultLitePullConsumer
;
import
org.apache.rocketmq.client.consumer.TopicMessageQueueChangeListener
;
import
org.apache.rocketmq.client.consumer.MessageSelector
;
import
org.apache.rocketmq.client.consumer.MessageQueueListener
;
import
org.apache.rocketmq.client.consumer.MessageSelector
;
import
org.apache.rocketmq.client.consumer.PullResult
;
import
org.apache.rocketmq.client.consumer.TopicMessageQueueChangeListener
;
import
org.apache.rocketmq.client.consumer.store.LocalFileOffsetStore
;
import
org.apache.rocketmq.client.consumer.store.OffsetStore
;
import
org.apache.rocketmq.client.consumer.store.ReadOffsetType
;
...
...
@@ -164,6 +163,10 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
throw
new
IllegalStateException
(
NOT_RUNNING_EXCEPTION_MESSAGE
);
}
public
void
updateNameServerAddr
(
String
newAddresses
)
{
this
.
mQClientFactory
.
getMQClientAPIImpl
().
updateNameServerAddressList
(
newAddresses
);
}
private
synchronized
void
setSubscriptionType
(
SubscriptionType
type
)
{
if
(
this
.
subscriptionType
==
SubscriptionType
.
NONE
)
this
.
subscriptionType
=
type
;
...
...
@@ -556,6 +559,16 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
}
}
public
void
seekToBegin
(
MessageQueue
messageQueue
)
throws
MQClientException
{
long
begin
=
minOffset
(
messageQueue
);
this
.
seek
(
messageQueue
,
begin
);
}
public
void
seekToEnd
(
MessageQueue
messageQueue
)
throws
MQClientException
{
long
begin
=
maxOffset
(
messageQueue
);
this
.
seek
(
messageQueue
,
begin
);
}
private
long
maxOffset
(
MessageQueue
messageQueue
)
throws
MQClientException
{
checkServiceState
();
return
this
.
mQClientFactory
.
getMQAdminImpl
().
maxOffset
(
messageQueue
);
...
...
@@ -764,8 +777,9 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
subscriptionData
=
FilterAPI
.
buildSubscriptionData
(
defaultLitePullConsumer
.
getConsumerGroup
(),
topic
,
SubscriptionData
.
SUB_ALL
);
}
PullResult
pullResult
=
pull
(
messageQueue
,
subscriptionData
,
offset
,
nextPullBatchSize
());
switch
(
pullResult
.
getPullStatus
())
{
case
FOUND:
final
Object
objLock
=
messageQueueLock
.
fetchLockObject
(
messageQueue
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录