/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * 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.rocketmq.example.quickstart;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.selector.Region;
import com.alibaba.rocketmq.client.producer.selector.SelectMessageQueueByRegion;
import com.alibaba.rocketmq.common.message.Message;
import com.google.common.util.concurrent.RateLimiter;

import java.util.Arrays;


/**
 * ExampleCacheableProducer，发送消息
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("PG_QuickStart");
        producer.setSendMsgTimeout(10000);
        producer.setSendLatencyFaultEnable(true);
        producer.start();
        byte[] data = new byte[1024];
        Arrays.fill(data, (byte)'x');
        float limit = 5000;
        if (args.length > 0) {
            limit = Float.parseFloat(args[0]);
        }
        Message msg = new Message("T_QuickStart",// topic
                "TagA",// tag
                data
        );
        msg.setWaitStoreMsgOK(false);
        RateLimiter rateLimiter = RateLimiter.create(limit);
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            rateLimiter.acquire();
            try {
//                SendResult sendResult = producer.send(msg);
//                System.out.println(sendResult);
                producer.send(msg, new SelectMessageQueueByRegion(Region.ANY), null, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                    }

                    @Override
                    public void onException(Throwable e) {
                        e.printStackTrace();
                    }
                }, 200, true);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        producer.shutdown();
    }
}

@@ -1181,16 +1181,194 @@
 Arrays;%0A
+import java.util.concurrent.Executors;%0Aimport java.util.concurrent.ScheduledExecutorService;%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.concurrent.atomic.AtomicLong;%0A
 %0A%0A/**%0A *
@@ -1816,9 +1816,10 @@
 t = 
-5
+12
 000;
@@ -2122,24 +2122,531 @@
 ate(limit);%0A
+        final AtomicLong success = new AtomicLong();%0A        final AtomicLong failure = new AtomicLong();%0A%0A        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();%0A        scheduledExecutorService.scheduleAtFixedRate(new Runnable() %7B%0A            @Override%0A            public void run() %7B%0A                System.out.println(%22S: %22 + success.get());%0A                System.out.println(%22F: %22 + failure.get());%0A            %7D%0A        %7D, 10, 1, TimeUnit.SECONDS);%0A%0A
         for 
@@ -3083,37 +3083,32 @@
     
-System.out.println(sendResult
+success.incrementAndGet(
 );%0A 
@@ -3242,33 +3242,39 @@
         
-e.printStackTrace
+failure.incrementAndGet
 ();%0A    
@@ -3310,17 +3310,18 @@
      %7D, 
-2
+60
 00, true

Add test