[RIP-19] Pop Consuming (add example & doc)

docs/en/Feature.md

@@ -84,3 +84,8 @@ The result of consumer flow control is to reduce the pull frequency.
 Dead letter queue is used to deal messages that cannot be consumed normally. When a message is consumed failed at first time, the message queue will automatically resend the message. If the consumption still fails after the maximum number retry, it indicates that the consumer cannot properly consume the message under normal circumstances. At this time, the message queue will not immediately abandon the message, but send it to the special queue corresponding to the consumer.
 
 RocketMQ defines the messages that could not be consumed under normal circumstances as Dead-Letter Messages, and the special queue in which the Dead-Letter Messages are saved as Dead-Letter Queues. In RocketMQ, the consumer instance can consume again by resending messages in the Dead-Letter Queue using console.
+
+## 13 Pop Consuming
+
+Pop consuming refers to that broker fetches messages from queues owned by same broker and returns to clients, which ensures one queue will be consumed by multiple clients. The whole behavior is like a queue
+pop process. By invoking `setConsumeMode` sub command of mqadmin, one consumer group can be switch to POP consuming instead of classical PULL consuming without changing a single code line. The new pop consuming will help to mitigate the impact for one queue consuming of an abnormal behaving client.
\ No newline at end of file

example/pom.xml

@@ -44,6 +44,10 @@
             <groupId>${project.groupId}</groupId>
             <artifactId>rocketmq-acl</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-tools</artifactId>
+        </dependency>
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>

example/src/main/java/org/apache/rocketmq/example/simple/PopPushConsumer.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.example.simple;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageRequestMode;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+
+public class PopPushConsumer {
+
+    public static final String CONSUMER_GROUP = "CID_JODIE_1";
+    public static final String TOPIC = "TopicTest";
+
+    // Or it can use cli directly: mqadmin setConsumeMode -c cluster -t topic -g group -m POP -n 8
+    private static void switchPop() throws Exception {
+        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
+        mqAdminExt.start();
+
+        ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+        Set<String> brokerAddrs = clusterInfo.getBrokerAddrTable().values().stream().map(BrokerData::selectBrokerAddr).collect(Collectors.toSet());
+
+        for (String brokerAddr : brokerAddrs) {
+            mqAdminExt.setMessageRequestMode(brokerAddr, TOPIC, CONSUMER_GROUP, MessageRequestMode.POP, 8, 3_000);
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        switchPop();
+
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
+        consumer.subscribe(TOPIC, "*");
+        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
+        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        consumer.start();
+        System.out.printf("Consumer Started.%n");
+    }
+}