[ISSUE #984] Support fifo parallel consume in java pushconsumer (#985)

Author: qianye1001

.github/workflows/java_build.yml

@@ -8,13 +8,13 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04, macos-latest, windows-2022]
+        os: [ ubuntu-22.04, macos-latest, windows-2022 ]
         jdk: [11, 17]
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Set up JDK ${{ matrix.jdk }}
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v3
         with:
           java-version: ${{ matrix.jdk }}
           distribution: "adopt"
@@ -27,10 +27,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout Current Repository
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       # Use JDK 17.
       - name: Use JDK 17
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v3
         with:
           java-version: 17
           distribution: "adopt"
@@ -53,7 +53,7 @@ jobs:
             | xargs echo "::set-output name=tag_name::"
       # Clone the opentelemetry-java-instrumentation repository.
       - name: Checkout Latest Release
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
         with:
           repository: open-telemetry/opentelemetry-java-instrumentation
           ref: ${{ steps.get_release_tag.outputs.tag_name }}
@@ -67,7 +67,7 @@ jobs:
           sed -i 's/org\.apache\.rocketmq:rocketmq-client-java:[^"]*/org.apache.rocketmq:rocketmq-client-java:${{ steps.get_version.outputs.version }}/' instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/testing/build.gradle.kts
       # Use JDK 17.
       - name: Use JDK 17
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v3
         with:
           java-version: 17
           distribution: "adopt"

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/PushConsumerBuilder.java

@@ -81,6 +81,15 @@ public interface PushConsumerBuilder {
      */
     PushConsumerBuilder setConsumptionThreadCount(int count);
 
+    /**
+     * Set enable fifo consume accelerator. If enabled, the consumer will consume messages in parallel by messageGroup,
+     * it may increase the probability of repeatedly consuming the same message.
+     *
+     * @param enableFifoConsumeAccelerator  enable fifo parallel processing.
+     * @return the consumer builder instance.
+     */
+    PushConsumerBuilder setEnableFifoConsumeAccelerator(boolean enableFifoConsumeAccelerator);
+
     /**
      * Finalize the build of {@link PushConsumer} and start.
      *

java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java

@@ -20,8 +20,12 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
@@ -35,16 +39,38 @@
 @SuppressWarnings("UnstableApiUsage")
 class FifoConsumeService extends ConsumeService {
     private static final Logger log = LoggerFactory.getLogger(FifoConsumeService.class);
+    private final boolean enableFifoConsumeAccelerator;
 
     public FifoConsumeService(ClientId clientId, MessageListener messageListener,
         ThreadPoolExecutor consumptionExecutor, MessageInterceptor messageInterceptor,
-        ScheduledExecutorService scheduler) {
+        ScheduledExecutorService scheduler, boolean enableFifoConsumeAccelerator) {
         super(clientId, messageListener, consumptionExecutor, messageInterceptor, scheduler);
+        this.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
     }
 
     @Override
     public void consume(ProcessQueue pq, List<MessageViewImpl> messageViews) {
-        consumeIteratively(pq, messageViews.iterator());
+        if (!enableFifoConsumeAccelerator || messageViews.size() <= 1) {
+            consumeIteratively(pq, messageViews.iterator());
+            return;
+        }
+        Map<String, List<MessageViewImpl>> messageViewsGroupByMessageGroup = new HashMap<>();
+        List<MessageViewImpl> messageViewsWithoutMessageGroup = new ArrayList<>();
+        for (MessageViewImpl messageView : messageViews) {
+            Optional<String> messageGroup = messageView.getMessageGroup();
+            if (messageGroup.isPresent()) {
+                messageViewsGroupByMessageGroup.computeIfAbsent(messageGroup.get(), k -> new ArrayList<>())
+                    .add(messageView);
+            } else {
+                messageViewsWithoutMessageGroup.add(messageView);
+            }
+        }
+
+        log.debug("FifoConsumeService parallel consume, messageViewsNum={}, groupNum={}", messageViews.size(),
+            messageViewsGroupByMessageGroup.size() + (messageViewsWithoutMessageGroup.isEmpty() ? 0 : 1));
+
+        messageViewsGroupByMessageGroup.values().forEach(list -> consumeIteratively(pq, list.iterator()));
+        consumeIteratively(pq, messageViewsWithoutMessageGroup.iterator());
     }
 
     public void consumeIteratively(ProcessQueue pq, Iterator<MessageViewImpl> iterator) {

java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerBuilderImpl.java

@@ -41,6 +41,7 @@ public class PushConsumerBuilderImpl implements PushConsumerBuilder {
     private int maxCacheMessageCount = 1024;
     private int maxCacheMessageSizeInBytes = 64 * 1024 * 1024;
     private int consumptionThreadCount = 20;
+    private boolean enableFifoConsumeAccelerator = false;
 
     /**
      * @see PushConsumerBuilder#setClientConfiguration(ClientConfiguration)
@@ -113,6 +114,15 @@ public PushConsumerBuilder setConsumptionThreadCount(int consumptionThreadCount)
         return this;
     }
 
+    /**
+     * @see PushConsumerBuilder#setEnableFifoConsumeAccelerator(boolean)
+     */
+    @Override
+    public PushConsumerBuilder setEnableFifoConsumeAccelerator(boolean enableFifoConsumeAccelerator) {
+        this.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
+        return this;
+    }
+
     /**
      * @see PushConsumerBuilder#build()
      */
@@ -124,7 +134,7 @@ public PushConsumer build() throws ClientException {
         checkArgument(!subscriptionExpressions.isEmpty(), "subscriptionExpressions have not been set yet");
         final PushConsumerImpl pushConsumer = new PushConsumerImpl(clientConfiguration, consumerGroup,
             subscriptionExpressions, messageListener, maxCacheMessageCount, maxCacheMessageSizeInBytes,
-            consumptionThreadCount);
+            consumptionThreadCount, enableFifoConsumeAccelerator);
         pushConsumer.startAsync().awaitRunning();
         return pushConsumer;
     }

java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java

@@ -102,6 +102,7 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer {
     private final MessageListener messageListener;
     private final int maxCacheMessageCount;
     private final int maxCacheMessageSizeInBytes;
+    private final boolean enableFifoConsumeAccelerator;
 
     /**
      * Indicates the times of message reception.
@@ -124,7 +125,8 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer {
      */
     public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumerGroup,
         Map<String, FilterExpression> subscriptionExpressions, MessageListener messageListener,
-        int maxCacheMessageCount, int maxCacheMessageSizeInBytes, int consumptionThreadCount) {
+        int maxCacheMessageCount, int maxCacheMessageSizeInBytes, int consumptionThreadCount,
+        boolean enableFifoConsumeAccelerator) {
         super(clientConfiguration, consumerGroup, subscriptionExpressions.keySet());
         this.clientConfiguration = clientConfiguration;
         Resource groupResource = new Resource(clientConfiguration.getNamespace(), consumerGroup);
@@ -136,6 +138,7 @@ public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumer
         this.messageListener = messageListener;
         this.maxCacheMessageCount = maxCacheMessageCount;
         this.maxCacheMessageSizeInBytes = maxCacheMessageSizeInBytes;
+        this.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
 
         this.receptionTimes = new AtomicLong(0);
         this.receivedMessagesQuantity = new AtomicLong(0);
@@ -153,6 +156,13 @@ public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumer
             new ThreadFactoryImpl("MessageConsumption", this.getClientId().getIndex()));
     }
 
+    public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumerGroup,
+        Map<String, FilterExpression> subscriptionExpressions, MessageListener messageListener,
+        int maxCacheMessageCount, int maxCacheMessageSizeInBytes, int consumptionThreadCount) {
+        this(clientConfiguration, consumerGroup, subscriptionExpressions, messageListener, maxCacheMessageCount,
+            maxCacheMessageSizeInBytes, consumptionThreadCount, true);
+    }
+
     @Override
     protected void startUp() throws Exception {
         try {
@@ -193,8 +203,10 @@ protected void shutDown() throws InterruptedException {
     private ConsumeService createConsumeService() {
         final ScheduledExecutorService scheduler = this.getClientManager().getScheduler();
         if (pushSubscriptionSettings.isFifo()) {
-            log.info("Create FIFO consume service, consumerGroup={}, clientId={}", consumerGroup, clientId);
-            return new FifoConsumeService(clientId, messageListener, consumptionExecutor, this, scheduler);
+            log.info("Create FIFO consume service, consumerGroup={}, clientId={}, enableFifoConsumeAccelerator={}",
+                consumerGroup, clientId, enableFifoConsumeAccelerator);
+            return new FifoConsumeService(clientId, messageListener, consumptionExecutor, this,
+                scheduler, enableFifoConsumeAccelerator);
         }
         log.info("Create standard consume service, consumerGroup={}, clientId={}", consumerGroup, clientId);
         return new StandardConsumeService(clientId, messageListener, consumptionExecutor, this, scheduler);