init

sentinel/sentinel-demo/sentinel-demo-rocketmq/README.md

@@ -0,0 +1,15 @@
+# Sentinel RocketMQ Demo
+
+This demonstrates some specific scenarios for Apache RocketMQ client.
+
+## Uniform Rate Limiting
+
+In Apache RocketMQ, when message consumers are consuming messages, there may a sudden inflow of messages, whether using pull or push mode. If all the messages were handled at this time, it would be likely to cause the system to be overloaded and then affect stability. However, in fact, there may be no messages coming within a few seconds. If redundant messages are directly discarded, the system's ability to process the message is not fully utilized. We hope that the sudden inflow of messages can be spread over a period of time, so that the system load can be kept on the stable level while processing as many messages as possible, thus achieving the effect of “shaving the peaks and filling the valley”.
+
+![shaving the peaks and filling the valley](https://github.com/alibaba/Sentinel/wiki/image/mq-traffic-peak-clipping-en.png)
+
+Sentinel provides a feature for this kind of scenario: [Rate Limiter](https://github.com/alibaba/Sentinel/wiki/Flow-Shaping:-Pace-Limiter), which can spread a large number of sudden request inflow in a uniform rate manner, let the request pass at a fixed interval. It is often used to process burst requests instead of rejecting them. This avoids traffic spurs causing system overloaded. Moreover, the pending requests will be queued and processed one by one. When the request is estimated to exceed the maximum queuing timeout, it will be rejected immediately.
+
+For example, we configure the rule with uniform rate limiting mode and QPS count is 5, which indicates messages are consumed at fixed interval (200 ms) and pending messages will queue. We also set the maximum queuing timeout is 5s, then all requests estimated to exceed the timeout will be rejected immediately.
+
+![Uniform rate](https://github.com/alibaba/Sentinel/wiki/image/uniform-speed-queue.png)

sentinel/sentinel-demo/sentinel-demo-rocketmq/pom.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>com.alibaba.csp</groupId>
+        <artifactId>sentinel-demo</artifactId>
+        <version>1.8.3</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-demo-rocketmq</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>4.2.0</version>
+        </dependency>
+    </dependencies>
+
+</project>

sentinel/sentinel-demo/sentinel-demo-rocketmq/src/main/java/com/alibaba/csp/sentinel/demo/rocketmq/Constants.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.csp.sentinel.demo.rocketmq;
+
+public final class Constants {
+
+    public static final String TEST_GROUP_NAME = "sentinel-group";
+    public static final String TEST_TOPIC_NAME = "SentinelTopicTest";
+    public static final String TEST_NAMESRV_ADDR = "127.0.0.1:9876";
+
+    private Constants() {}
+}

sentinel/sentinel-demo/sentinel-demo-rocketmq/src/main/java/com/alibaba/csp/sentinel/demo/rocketmq/PullConsumerDemo.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.csp.sentinel.demo.rocketmq;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.alibaba.csp.sentinel.Entry;
+import com.alibaba.csp.sentinel.EntryType;
+import com.alibaba.csp.sentinel.SphU;
+import com.alibaba.csp.sentinel.context.ContextUtil;
+import com.alibaba.csp.sentinel.slots.block.BlockException;
+import com.alibaba.csp.sentinel.slots.block.RuleConstant;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
+import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+
+public class PullConsumerDemo {
+
+    private static final String KEY = String.format("%s:%s", Constants.TEST_GROUP_NAME, Constants.TEST_TOPIC_NAME);
+
+    private static final Map<MessageQueue, Long> OFFSET_TABLE = new HashMap<MessageQueue, Long>();
+
+    @SuppressWarnings("PMD.ThreadPoolCreationRule")
+    private static final ExecutorService pool = Executors.newFixedThreadPool(32);
+
+    private static final AtomicLong SUCCESS_COUNT = new AtomicLong(0);
+    private static final AtomicLong FAIL_COUNT = new AtomicLong(0);
+
+    public static void main(String[] args) throws MQClientException {
+        // First we init the flow control rule for Sentinel.
+        initFlowControlRule();
+
+        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(Constants.TEST_GROUP_NAME);
+        consumer.setNamesrvAddr(Constants.TEST_NAMESRV_ADDR);
+        consumer.start();
+
+        Set<MessageQueue> mqs = new HashSet<>();
+        try {
+            mqs = consumer.fetchSubscribeMessageQueues(Constants.TEST_TOPIC_NAME);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        for (MessageQueue mq : mqs) {
+            System.out.printf("Consuming messages from the queue: %s%n", mq);
+            SINGLE_MQ:
+            while (true) {
+                try {
+                    PullResult pullResult =
+                        consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
+                    if (pullResult.getMsgFoundList() != null) {
+                        for (MessageExt msg : pullResult.getMsgFoundList()) {
+                            doSomething(msg);
+                        }
+                    }
+
+                    long nextOffset = pullResult.getNextBeginOffset();
+                    putMessageQueueOffset(mq, nextOffset);
+                    consumer.updateConsumeOffset(mq, nextOffset);
+                    switch (pullResult.getPullStatus()) {
+                        case NO_NEW_MSG:
+                            break SINGLE_MQ;
+                        case FOUND:
+                        case NO_MATCHED_MSG:
+                        case OFFSET_ILLEGAL:
+                        default:
+                            break;
+                    }
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+
+        consumer.shutdown();
+    }
+
+    private static void doSomething(MessageExt message) {
+        pool.submit(() -> {
+            Entry entry = null;
+            try {
+                ContextUtil.enter(KEY);
+                entry = SphU.entry(KEY, EntryType.OUT);
+
+                // Your business logic here.
+                System.out.printf("[%d][%s][Success: %d] Receive New Messages: %s %n", System.currentTimeMillis(),
+                    Thread.currentThread().getName(), SUCCESS_COUNT.addAndGet(1), new String(message.getBody()));
+            } catch (BlockException ex) {
+                // Blocked.
+                System.out.println("Blocked: " + FAIL_COUNT.addAndGet(1));
+            } finally {
+                if (entry != null) {
+                    entry.exit();
+                }
+                ContextUtil.exit();
+            }
+        });
+    }
+
+    private static void initFlowControlRule() {
+        FlowRule rule = new FlowRule();
+        rule.setResource(KEY);
+        // Indicates the interval between two adjacent requests is 200 ms.
+        rule.setCount(5);
+        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
+        rule.setLimitApp("default");
+
+        // Enable rate limiting (uniform). This can ensure fixed intervals between two adjacent calls.
+        // In this example, intervals between two incoming calls (message consumption) will be 200 ms constantly.
+        rule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER);
+        // If more requests are coming, they'll be put into the waiting queue.
+        // The queue has a queueing timeout. Requests that may exceed the timeout will be immediately blocked.
+        // In this example, the max timeout is 5s.
+        rule.setMaxQueueingTimeMs(5 * 1000);
+        FlowRuleManager.loadRules(Collections.singletonList(rule));
+    }
+
+    private static long getMessageQueueOffset(MessageQueue mq) {
+        Long offset = OFFSET_TABLE.get(mq);
+        if (offset != null) {
+            return offset;
+        }
+
+        return 0;
+    }
+
+    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
+        OFFSET_TABLE.put(mq, offset);
+    }
+
+}

sentinel/sentinel-demo/sentinel-demo-rocketmq/src/main/java/com/alibaba/csp/sentinel/demo/rocketmq/SyncProducer.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.csp.sentinel.demo.rocketmq;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+public class SyncProducer {
+
+    public static void main(String[] args) throws Exception {
+        // Instantiate with a producer group name.
+        DefaultMQProducer producer = new DefaultMQProducer(Constants.TEST_GROUP_NAME);
+        producer.setNamesrvAddr(Constants.TEST_NAMESRV_ADDR);
+        // Launch the instance.
+        producer.start();
+        for (int i = 0; i < 1000; i++) {
+            // Create a message instance, specifying topic, tag and message body.
+            Message msg = new Message(Constants.TEST_TOPIC_NAME, "TagA",
+                ("Hello RocketMQ From Sentinel " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
+            );
+
+            try {
+                // Call send message to deliver message to one of brokers.
+                SendResult sendResult = producer.send(msg);
+                System.out.printf("%s%n", sendResult);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        // Shut down once the producer instance is not longer in use.
+        producer.shutdown();
+    }
+}