drpmma commented on code in PR #6278:
URL: https://github.com/apache/rocketmq/pull/6278#discussion_r1129040461
##########
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java:
##########
@@ -1021,16 +1032,154 @@ public MessageQueue getMessageQueue() {
}
}
+ public void pullMessage(final MessageQueue messageQueue) {
Review Comment:
Is it possible to abstract a method for the purpose of code reuse rather
than duplicating it?
https://github.com/apache/rocketmq/blob/a3228ad2736c0311592b12f994361b06cf4f72a9/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java#L896-L1020
##########
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java:
##########
@@ -131,9 +128,17 @@ private enum SubscriptionType {
private DefaultLitePullConsumer defaultLitePullConsumer;
+ private PullMessageQueueService pullMessageQueueService;
+
private final ConcurrentMap<MessageQueue, PullTaskImpl> taskTable =
new ConcurrentHashMap<>();
+ // Dummy value to associate with an Object in the backing Map
+ private static final Object PRESENT = new Object();
+
+ private final ConcurrentMap<MessageQueue, Object> messageQueueTable =
Review Comment:
It's quite strange to use Object as the map value. Are you attempting to
create a concurrent set? If that is the case, you may use the keySet of a
ConcurrentMap instead.
##########
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java:
##########
@@ -450,20 +462,22 @@ public PullAPIWrapper getPullAPIWrapper() {
private void startPullTask(Collection<MessageQueue> mqSet) {
for (MessageQueue messageQueue : mqSet) {
- if (!this.taskTable.containsKey(messageQueue)) {
- PullTaskImpl pullTask = new PullTaskImpl(messageQueue);
- this.taskTable.put(messageQueue, pullTask);
- this.scheduledThreadPoolExecutor.schedule(pullTask, 0,
TimeUnit.MILLISECONDS);
+ if (!this.messageQueueTable.containsKey(messageQueue)) {
+// PullTaskImpl pullTask = new PullTaskImpl(messageQueue);
Review Comment:
It's suggested to remove the comment line.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
