You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "RongtongJin (via GitHub)" <gi...@apache.org> on 2023/04/10 09:40:00 UTC
[GitHub] [rocketmq] RongtongJin commented on a diff in pull request #6417: [ISSUE #6205] Support Async pull for DefaultLitePullConsumer
RongtongJin commented on code in PR #6417:
URL: https://github.com/apache/rocketmq/pull/6417#discussion_r1154175155
##########
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java:
##########
@@ -962,43 +972,83 @@ public void run() {
subscriptionData = FilterAPI.buildSubscriptionData(topic, subExpression4Assign);
}
- PullResult pullResult = pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize());
- if (this.isCancelled() || processQueue.isDropped()) {
- return;
- }
- switch (pullResult.getPullStatus()) {
- case FOUND:
- final Object objLock = messageQueueLock.fetchLockObject(messageQueue);
- synchronized (objLock) {
- if (pullResult.getMsgFoundList() != null && !pullResult.getMsgFoundList().isEmpty() && assignedMessageQueue.getSeekOffset(messageQueue) == -1) {
- processQueue.putMessage(pullResult.getMsgFoundList());
- submitConsumeRequest(new ConsumeRequest(pullResult.getMsgFoundList(), messageQueue, processQueue));
+ PullResult pullResult;
+ switch (communicationMode) {
+ case ASYNC:
+ PullCallback pullCallback = new PullCallback() {
+ @Override
+ public void onSuccess(PullResult pullResult) {
+ addResultToQueue(pullResult, processQueue);
+ executeThreadWithDelay();
}
- }
- break;
- case OFFSET_ILLEGAL:
- log.warn("The pull request offset illegal, {}", pullResult.toString());
+
+ @Override
+ public void onException(Throwable e) {
+ handleError(e);
+ executeThreadWithDelay();
+ }
+ };
+ pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize(), pullCallback);
+ return;
+ case SYNC:
+ pullResult = pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize());
+
+ addResultToQueue(pullResult, processQueue);
break;
default:
+ assert false;
break;
}
- updatePullOffset(messageQueue, pullResult.getNextBeginOffset(), processQueue);
- } catch (InterruptedException interruptedException) {
- log.warn("Polling thread was interrupted.", interruptedException);
- } catch (Throwable e) {
- if (e instanceof MQBrokerException && ((MQBrokerException) e).getResponseCode() == ResponseCode.FLOW_CONTROL) {
- pullDelayTimeMills = PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL;
- } else {
- pullDelayTimeMills = pullTimeDelayMillsWhenException;
- }
- log.error("An error occurred in pull message process.", e);
+ } catch (Exception e) {
+ // For pull() sync and FilterAPI.buildSubscriptionData
+ handleError(e);
}
+ executeThreadWithDelay();
+ }
+ }
- if (!this.isCancelled()) {
- scheduledThreadPoolExecutor.schedule(this, pullDelayTimeMills, TimeUnit.MILLISECONDS);
+ private void addResultToQueue(PullResult pullResult, ProcessQueue processQueue) {
+ if (this.isCancelled() || processQueue.isDropped()) {
+ return;
+ }
+
+ switch (pullResult.getPullStatus()) {
+ case FOUND:
+ final Object objLock = messageQueueLock.fetchLockObject(messageQueue);
+ synchronized (objLock) {
+ if (pullResult.getMsgFoundList() != null && !pullResult.getMsgFoundList().isEmpty() && assignedMessageQueue.getSeekOffset(messageQueue) == -1) {
+ processQueue.putMessage(pullResult.getMsgFoundList());
+ submitConsumeRequest(new ConsumeRequest(pullResult.getMsgFoundList(), messageQueue, processQueue));
+ }
+ }
+ break;
+ case OFFSET_ILLEGAL:
+ log.warn("The pull request offset illegal, {}", pullResult.toString());
+ break;
+ default:
+ break;
+ }
+ updatePullOffset(messageQueue, pullResult.getNextBeginOffset(), processQueue);
+ }
+
+ private void handleError(Throwable e) {
+ if (e instanceof InterruptedException) {
+ log.warn("Polling thread was interrupted.", e);
+ } else {
+ if (e instanceof MQBrokerException && ((MQBrokerException) e).getResponseCode() == ResponseCode.FLOW_CONTROL) {
+ pullDelayTimeMills = PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL;
} else {
- log.warn("The Pull Task is cancelled after doPullTask, {}", messageQueue);
+ pullDelayTimeMills = pullTimeDelayMillsWhenException;
}
+ log.error("An error occurred in pull message process.", e);
+ }
+ }
+
+ private void executeThreadWithDelay() {
Review Comment:
How about rename to schedulePullTaskWithDelay？
##########
example/src/main/java/org/apache/rocketmq/example/simple/LitePullConsumerAsync.java:
##########
@@ -0,0 +1,42 @@
+package org.apache.rocketmq.example.simple;
Review Comment:
IMO, no need for a new example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org