You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by sh...@apache.org on 2023/07/30 13:49:34 UTC
[rocketmq-eventbridge] 01/02: update rocketmq consumer pull logic
This is an automated email from the ASF dual-hosted git repository.
shenlin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-eventbridge.git
commit f1a50397695c2b878235d589a7d1f130c1755f37
Author: changfeng <yo...@alibaba-inc.com>
AuthorDate: Sun Jul 30 11:35:34 2023 +0800
update rocketmq consumer pull logic
---
.../rocketmq/runtimer/consumer/LitePullConsumerImpl.java | 13 +++----------
.../rocketmq/runtimer/consumer/LocalMessageCache.java | 8 +++++---
2 files changed, 8 insertions(+), 13 deletions(-)
diff --git a/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LitePullConsumerImpl.java b/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LitePullConsumerImpl.java
index 7636449..a76012e 100644
--- a/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LitePullConsumerImpl.java
+++ b/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LitePullConsumerImpl.java
@@ -63,6 +63,7 @@ public class LitePullConsumerImpl implements LitePullConsumer {
private static final Long PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL = 30L;
private static final Long PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION = TimeUnit.SECONDS.toMillis(3);
private static final String DEFAULT_INSTANCE_NAME = "EventBridge_Consumer_INSTANCE";
+ private static final Integer PULL_BATCH_NUM = 32;
public LitePullConsumerImpl(final ClientConfig clientConfig, final RPCHook rpcHook) {
this.clientConfig = clientConfig;
@@ -224,16 +225,8 @@ public class LitePullConsumerImpl implements LitePullConsumer {
return;
}
long offset = localMessageCache.nextPullOffset(messageQueue);
- int batchNums = localMessageCache.nextPullBatchNums();
- // If batchNums is zero, an exception will be thrown and then trigger a delay
- if (batchNums <= 0) {
- log.warn("Local cache is full, delay the pull task {} ms for message queue {}",
- PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION, messageQueue);
- pullLater(PullTask.this, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION, TimeUnit.MILLISECONDS);
- return;
- }
- rocketmqPullConsumer.pullBlockIfNotFound(this.messageQueue, this.tag, offset, batchNums, new PullCallback() {
+ rocketmqPullConsumer.pullBlockIfNotFound(this.messageQueue, this.tag, offset, PULL_BATCH_NUM, new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
try {
@@ -249,7 +242,7 @@ public class LitePullConsumerImpl implements LitePullConsumer {
if (pq != null && !pq.isDropped()) {
pq.putMessage(pullResult.getMsgFoundList());
for (final MessageExt messageExt : pullResult.getMsgFoundList()) {
- localMessageCache.submitConsumeRequest(new ConsumeRequest(messageExt, messageQueue, pq));
+ localMessageCache.submitConsumeRequest(new ConsumeRequest(messageExt, messageQueue, pq), Long.MAX_VALUE);
}
localMessageCache.updatePullOffset(messageQueue, pullResult.getNextBeginOffset());
pullImmediately(PullTask.this);
diff --git a/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LocalMessageCache.java b/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LocalMessageCache.java
index 7d1c1af..f3e3617 100644
--- a/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LocalMessageCache.java
+++ b/adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/consumer/LocalMessageCache.java
@@ -124,11 +124,13 @@ public class LocalMessageCache {
pullOffsetTable.entrySet().removeIf(next -> !mqDivided.contains(next.getKey()));
}
- void submitConsumeRequest(ConsumeRequest consumeRequest) {
+ boolean submitConsumeRequest(ConsumeRequest consumeRequest, Long timeout) {
try {
- consumeRequestCache.put(consumeRequest);
- } catch (InterruptedException ignore) {
+ return consumeRequestCache.offer(consumeRequest, timeout, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.warn("put consumeRequestCache failed", e);
}
+ return false;
}
public List<MessageExt> poll(final int pullBatchSize, final Duration timeout) {